Non-replayable WAL records through overflows and >MaxAllocSize lengths

From 2da7f0258e458bf37df79ca3efcf2bd37b27e667 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Fri, 11 Mar 2022 16:16:22 +0100
Subject: [PATCH v2] Add protections in xlog record APIs against large numbers
 and overflows.

Before this; it was possible for an extension to create malicious WAL records
that were too large to replay; or that would overflow the xl_tot_len field,
causing potential corruption in WAL record IO ops.
---
 src/backend/access/transam/xloginsert.c | 37 +++++++++++++++++++++----
 src/include/access/xlogrecord.h         |  4 +++
 2 files changed, 35 insertions(+), 6 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index c260310c4c..71a649e216 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -338,10 +338,16 @@ XLogRegisterData(char *data, int len)
 {
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len >= 0 && AllocSizeIsValid(len));
 
-	if (num_rdatas >= max_rdatas)
+	/*
+	 * Check against max_rdatas; and ensure we don't fill a record with
+	 * more data than can be replayed
+	 */
+	if (unlikely(num_rdatas >= max_rdatas ||
+				 !AllocSizeIsValid((uint64) mainrdata_len + (uint64) len)))
 		elog(ERROR, "too much WAL data");
+
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -377,7 +383,7 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len >= 0 && len <= UINT16_MAX);
 
 	/* find the registered buffer struct */
 	regbuf = &registered_buffers[block_id];
@@ -385,8 +391,14 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 		elog(ERROR, "no block with id %d registered with WAL insertion",
 			 block_id);
 
-	if (num_rdatas >= max_rdatas)
+	/*
+	 * Check against max_rdatas; and ensure we don't register more data per
+	 * buffer than can be handled by the physical record format.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas ||
+				 regbuf->rdata_len + len > UINT16_MAX))
 		elog(ERROR, "too much WAL data");
+
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -505,7 +517,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, bool *topxid_included)
 {
 	XLogRecData *rdt;
-	uint32		total_len = 0;
+	uint64		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
@@ -734,12 +746,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 
 		if (needs_data)
 		{
+			/*
+			 * When copying to XLogRecordBlockHeader, the length is narrowed
+			 * to an uint16. We double-check that that is still correct.
+			 */
+			Assert(regbuf->rdata_len <= UINT16_MAX);
+
 			/*
 			 * Link the caller-supplied rdata chain for this buffer to the
 			 * overall list.
 			 */
 			bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
-			bkpb.data_length = regbuf->rdata_len;
+			bkpb.data_length = (uint16) regbuf->rdata_len;
 			total_len += regbuf->rdata_len;
 
 			rdt_datas_last->next = regbuf->rdata_head;
@@ -836,6 +854,13 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that xlogreader.c can read the record by ensuring that the
+	 * data section of the WAL record can be allocated.
+	 */
+	if (unlikely(!AllocSizeIsValid(total_len)))
+		elog(ERROR, "too much registered data for WAL record");
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index c1b1137aa7..950e1f22b0 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -37,6 +37,10 @@
  * The XLogRecordBlockHeader, XLogRecordDataHeaderShort and
  * XLogRecordDataHeaderLong structs all begin with a single 'id' byte. It's
  * used to distinguish between block references, and the main data structs.
+ *
+ * Note that xlogreader.c limits the total size of the xl_tot_len to
+ * MaxAllocSize (1GB - 1). This means that this is also the maximum size
+ * of a single WAL record - we would be unable to replay any larger record.
  */
 typedef struct XLogRecord
 {
-- 
2.30.2

From 9e4b70ca096469616fd1320a02cbde129a4943b6 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Fri, 11 Mar 2022 16:16:22 +0100
Subject: [PATCH v3] Add protections in xlog record APIs against large numbers
 and overflows.

Before this; it was possible for an extension to create malicious WAL records
that were too large to replay; or that would overflow the xl_tot_len field,
causing potential corruption in WAL record IO ops.
---
 src/backend/access/transam/xloginsert.c | 56 ++++++++++++++++++++-----
 src/include/access/xloginsert.h         |  4 +-
 src/include/access/xlogrecord.h         |  4 ++
 3 files changed, 52 insertions(+), 12 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index c260310c4c..e87642ccae 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -127,6 +127,17 @@ static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info,
 									   bool *topxid_included);
 static bool XLogCompressBackupBlock(char *page, uint16 hole_offset,
 									uint16 hole_length, char *dest, uint16 *dlen);
+static inline void XLogErrorDataLimitExceeded();
+
+/*
+ * Error due to exceeding the maximum size of a WAL record, or registering
+ * more datas than are being accounted for by the XLog infrastructure.
+ */
+inline void
+XLogErrorDataLimitExceeded()
+{
+	elog(ERROR, "too much WAL data");
+}
 
 /*
  * Begin constructing a WAL record. This must be called before the
@@ -334,14 +345,20 @@ XLogRegisterBlock(uint8 block_id, RelFileNode *rnode, ForkNumber forknum,
  * XLogRecGetData().
  */
 void
-XLogRegisterData(char *data, int len)
+XLogRegisterData(char *data, uint32 len)
 {
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && AllocSizeIsValid(len));
+
+	/*
+	 * Check against max_rdatas; and ensure we don't fill a record with
+	 * more data than can be replayed
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(!AllocSizeIsValid((uint64) mainrdata_len + (uint64) len)))
+		XLogErrorDataLimitExceeded();
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -372,12 +389,12 @@ XLogRegisterData(char *data, int len)
  * limited)
  */
 void
-XLogRegisterBufData(uint8 block_id, char *data, int len)
+XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 {
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len <= UINT16_MAX);
 
 	/* find the registered buffer struct */
 	regbuf = &registered_buffers[block_id];
@@ -385,8 +402,14 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 		elog(ERROR, "no block with id %d registered with WAL insertion",
 			 block_id);
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
+	/*
+	 * Check against max_rdatas; and ensure we don't register more data per
+	 * buffer than can be handled by the physical record format.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(regbuf->rdata_len + len > UINT16_MAX))
+		XLogErrorDataLimitExceeded();
+
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -505,7 +528,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, bool *topxid_included)
 {
 	XLogRecData *rdt;
-	uint32		total_len = 0;
+	uint64		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
@@ -734,12 +757,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 
 		if (needs_data)
 		{
+			/*
+			 * When copying to XLogRecordBlockHeader, the length is narrowed
+			 * to an uint16. We double-check that that is still correct.
+			 */
+			Assert(regbuf->rdata_len <= UINT16_MAX);
+
 			/*
 			 * Link the caller-supplied rdata chain for this buffer to the
 			 * overall list.
 			 */
 			bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
-			bkpb.data_length = regbuf->rdata_len;
+			bkpb.data_length = (uint16) regbuf->rdata_len;
 			total_len += regbuf->rdata_len;
 
 			rdt_datas_last->next = regbuf->rdata_head;
@@ -836,6 +865,13 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that xlogreader.c can read the record by ensuring that the
+	 * data section of the WAL record can be allocated.
+	 */
+	if (unlikely(!AllocSizeIsValid(total_len)))
+		XLogErrorDataLimitExceeded();
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
diff --git a/src/include/access/xloginsert.h b/src/include/access/xloginsert.h
index 5fc340c434..5ca0b0c2f0 100644
--- a/src/include/access/xloginsert.h
+++ b/src/include/access/xloginsert.h
@@ -43,12 +43,12 @@ extern void XLogBeginInsert(void);
 extern void XLogSetRecordFlags(uint8 flags);
 extern XLogRecPtr XLogInsert(RmgrId rmid, uint8 info);
 extern void XLogEnsureRecordSpace(int max_block_id, int ndatas);
-extern void XLogRegisterData(char *data, int len);
+extern void XLogRegisterData(char *data, uint32 len);
 extern void XLogRegisterBuffer(uint8 block_id, Buffer buffer, uint8 flags);
 extern void XLogRegisterBlock(uint8 block_id, RelFileNode *rnode,
 							  ForkNumber forknum, BlockNumber blknum, char *page,
 							  uint8 flags);
-extern void XLogRegisterBufData(uint8 block_id, char *data, int len);
+extern void XLogRegisterBufData(uint8 block_id, char *data, uint32 len);
 extern void XLogResetInsertion(void);
 extern bool XLogCheckBufferNeedsBackup(Buffer buffer);
 
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index c1b1137aa7..950e1f22b0 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -37,6 +37,10 @@
  * The XLogRecordBlockHeader, XLogRecordDataHeaderShort and
  * XLogRecordDataHeaderLong structs all begin with a single 'id' byte. It's
  * used to distinguish between block references, and the main data structs.
+ *
+ * Note that xlogreader.c limits the total size of the xl_tot_len to
+ * MaxAllocSize (1GB - 1). This means that this is also the maximum size
+ * of a single WAL record - we would be unable to replay any larger record.
  */
 typedef struct XLogRecord
 {
-- 
2.30.2

From f4cc40351f56fdb6eaa4531c85677e80ea6feaf5 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Mon, 20 Jun 2022 10:21:22 +0200
Subject: [PATCH v4] Add protections in xlog record APIs against large numbers
 and overflows.

Before this; it was possible for an extension to create malicious WAL records
that were too large to replay; or that would overflow the xl_tot_len field,
causing potential corruption in WAL record IO ops. Emitting invalid records
was also possible through pg_logical_emit_message(), which allowed you to emit
arbitrary amounts of data up to 2GB, much higher than the replay limit of 1GB.
---
 src/backend/access/transam/xloginsert.c | 62 +++++++++++++++++++++----
 src/include/access/xloginsert.h         |  4 +-
 src/include/access/xlogrecord.h         |  4 ++
 3 files changed, 58 insertions(+), 12 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 2ce9be2cc7..a3387a9265 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -141,7 +141,18 @@ static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info,
 									   bool *topxid_included);
 static bool XLogCompressBackupBlock(char *page, uint16 hole_offset,
 									uint16 hole_length, char *dest, uint16 *dlen);
+static inline void XLogErrorDataLimitExceeded();
 
+/*
+ * Error due to exceeding the maximum size of a WAL record, or registering
+ * more datas than are being accounted for by the XLog infrastructure.
+ */
+static inline void
+XLogErrorDataLimitExceeded()
+{
+	elog(ERROR, "too much WAL data");
+}
+ 
 /*
  * Begin constructing a WAL record. This must be called before the
  * XLogRegister* functions and XLogInsert().
@@ -348,14 +359,24 @@ XLogRegisterBlock(uint8 block_id, RelFileNode *rnode, ForkNumber forknum,
  * XLogRecGetData().
  */
 void
-XLogRegisterData(char *data, int len)
+XLogRegisterData(char *data, uint32 len)
 {
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && AllocSizeIsValid(len));
+
+	/*
+	 * Check against max_rdatas; and ensure we don't fill a record with
+	 * more data than can be replayed. Records are allocated in one chunk, so
+	 * ensure AllocSizeIsValid() for that size of record.
+	 *
+	 * We also protect against overflows by first casting to uint64 - with only
+	 * uint32-operations this could still be an issue.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(!AllocSizeIsValid((uint64) mainrdata_len + (uint64) len)))
+		XLogErrorDataLimitExceeded();
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -386,12 +407,12 @@ XLogRegisterData(char *data, int len)
  * limited)
  */
 void
-XLogRegisterBufData(uint8 block_id, char *data, int len)
+XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 {
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len <= UINT16_MAX);
 
 	/* find the registered buffer struct */
 	regbuf = &registered_buffers[block_id];
@@ -399,8 +420,16 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 		elog(ERROR, "no block with id %d registered with WAL insertion",
 			 block_id);
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
+	/*
+	 * Check against max_rdatas; and ensure we don't register more data per
+	 * buffer than can be handled by the physical data format; 
+	 * i.e. that regbuf->rdata_len does not grow beyond what
+	 * XLogRecordBlockHeader->data_length can hold.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(regbuf->rdata_len + len > UINT16_MAX))
+		XLogErrorDataLimitExceeded();
+
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -519,7 +548,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, bool *topxid_included)
 {
 	XLogRecData *rdt;
-	uint32		total_len = 0;
+	uint64		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
@@ -756,12 +785,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 
 		if (needs_data)
 		{
+			/*
+			 * When copying to XLogRecordBlockHeader, the length is narrowed
+			 * to an uint16. We double-check that that is still correct.
+			 */
+			Assert(regbuf->rdata_len <= UINT16_MAX);
+
 			/*
 			 * Link the caller-supplied rdata chain for this buffer to the
 			 * overall list.
 			 */
 			bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
-			bkpb.data_length = regbuf->rdata_len;
+			bkpb.data_length = (uint16) regbuf->rdata_len;
 			total_len += regbuf->rdata_len;
 
 			rdt_datas_last->next = regbuf->rdata_head;
@@ -858,6 +893,13 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that xlogreader.c can read the record by ensuring that the
+	 * data section of the WAL record can be allocated.
+	 */
+	if (unlikely(!AllocSizeIsValid(total_len)))
+		XLogErrorDataLimitExceeded();
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
diff --git a/src/include/access/xloginsert.h b/src/include/access/xloginsert.h
index 5fc340c434..5ca0b0c2f0 100644
--- a/src/include/access/xloginsert.h
+++ b/src/include/access/xloginsert.h
@@ -43,12 +43,12 @@ extern void XLogBeginInsert(void);
 extern void XLogSetRecordFlags(uint8 flags);
 extern XLogRecPtr XLogInsert(RmgrId rmid, uint8 info);
 extern void XLogEnsureRecordSpace(int max_block_id, int ndatas);
-extern void XLogRegisterData(char *data, int len);
+extern void XLogRegisterData(char *data, uint32 len);
 extern void XLogRegisterBuffer(uint8 block_id, Buffer buffer, uint8 flags);
 extern void XLogRegisterBlock(uint8 block_id, RelFileNode *rnode,
 							  ForkNumber forknum, BlockNumber blknum, char *page,
 							  uint8 flags);
-extern void XLogRegisterBufData(uint8 block_id, char *data, int len);
+extern void XLogRegisterBufData(uint8 block_id, char *data, uint32 len);
 extern void XLogResetInsertion(void);
 extern bool XLogCheckBufferNeedsBackup(Buffer buffer);
 
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 052ac6817a..3575363e84 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -37,6 +37,10 @@
  * The XLogRecordBlockHeader, XLogRecordDataHeaderShort and
  * XLogRecordDataHeaderLong structs all begin with a single 'id' byte. It's
  * used to distinguish between block references, and the main data structs.
+ *
+ * Note that xlogreader.c limits the total size of the xl_tot_len to
+ * MaxAllocSize (1GB - 1). This means that this is also the maximum size
+ * of a single WAL record - we would be unable to replay any larger record.
  */
 typedef struct XLogRecord
 {
-- 
2.30.2

From e980e58d9cf8470a6ed12b0d8ac979c68d5cc17c Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Mon, 20 Jun 2022 10:21:22 +0200
Subject: [PATCH v4] Add protections in xlog record APIs against large numbers
 and overflows.

Before this; it was possible for an extension to create malicious WAL records
that were too large to replay; or that would overflow the xl_tot_len field,
causing potential corruption in WAL record IO ops. Emitting invalid records
was also possible through pg_logical_emit_message(), which allowed you to emit
arbitrary amounts of data up to 2GB, much higher than the replay limit of 1GB.
---
 src/backend/access/transam/xloginsert.c | 58 +++++++++++++++++++++----
 src/include/access/xlogrecord.h         |  4 ++
 2 files changed, 54 insertions(+), 8 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 2ce9be2cc7..0308531096 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -141,7 +141,18 @@ static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info,
 									   bool *topxid_included);
 static bool XLogCompressBackupBlock(char *page, uint16 hole_offset,
 									uint16 hole_length, char *dest, uint16 *dlen);
+static inline void XLogErrorDataLimitExceeded();
 
+/*
+ * Error due to exceeding the maximum size of a WAL record, or registering
+ * more datas than are being accounted for by the XLog infrastructure.
+ */
+static inline void
+XLogErrorDataLimitExceeded()
+{
+	elog(ERROR, "too much WAL data");
+}
+ 
 /*
  * Begin constructing a WAL record. This must be called before the
  * XLogRegister* functions and XLogInsert().
@@ -352,10 +363,20 @@ XLogRegisterData(char *data, int len)
 {
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && AllocSizeIsValid(len) && len >= 0);
+
+	/*
+	 * Check against max_rdatas; and ensure we don't fill a record with
+	 * more data than can be replayed. Records are allocated in one chunk, so
+	 * ensure AllocSizeIsValid() for that size of record.
+	 *
+	 * We also protect against overflows by first casting to uint64 - with only
+	 * uint32-operations this could still be an issue.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(!AllocSizeIsValid((uint64) mainrdata_len + (uint64) len)))
+		XLogErrorDataLimitExceeded();
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -391,7 +412,7 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len <= UINT16_MAX && len >= 0);
 
 	/* find the registered buffer struct */
 	regbuf = &registered_buffers[block_id];
@@ -399,8 +420,16 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 		elog(ERROR, "no block with id %d registered with WAL insertion",
 			 block_id);
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
+	/*
+	 * Check against max_rdatas; and ensure we don't register more data per
+	 * buffer than can be handled by the physical data format; 
+	 * i.e. that regbuf->rdata_len does not grow beyond what
+	 * XLogRecordBlockHeader->data_length can hold.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(regbuf->rdata_len + len > UINT16_MAX))
+		XLogErrorDataLimitExceeded();
+
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -519,7 +548,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, bool *topxid_included)
 {
 	XLogRecData *rdt;
-	uint32		total_len = 0;
+	uint64		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
@@ -756,12 +785,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 
 		if (needs_data)
 		{
+			/*
+			 * When copying to XLogRecordBlockHeader, the length is narrowed
+			 * to an uint16. We double-check that that is still correct.
+			 */
+			Assert(regbuf->rdata_len <= UINT16_MAX);
+
 			/*
 			 * Link the caller-supplied rdata chain for this buffer to the
 			 * overall list.
 			 */
 			bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
-			bkpb.data_length = regbuf->rdata_len;
+			bkpb.data_length = (uint16) regbuf->rdata_len;
 			total_len += regbuf->rdata_len;
 
 			rdt_datas_last->next = regbuf->rdata_head;
@@ -858,6 +893,13 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that xlogreader.c can read the record by ensuring that the
+	 * data section of the WAL record can be allocated.
+	 */
+	if (unlikely(!AllocSizeIsValid(total_len)))
+		XLogErrorDataLimitExceeded();
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 052ac6817a..3575363e84 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -37,6 +37,10 @@
  * The XLogRecordBlockHeader, XLogRecordDataHeaderShort and
  * XLogRecordDataHeaderLong structs all begin with a single 'id' byte. It's
  * used to distinguish between block references, and the main data structs.
+ *
+ * Note that xlogreader.c limits the total size of the xl_tot_len to
+ * MaxAllocSize (1GB - 1). This means that this is also the maximum size
+ * of a single WAL record - we would be unable to replay any larger record.
  */
 typedef struct XLogRecord
 {
-- 
2.30.2

From a9647e8ef3f3e8036e051a186818888dd8f65a80 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Mon, 20 Jun 2022 10:21:22 +0200
Subject: [PATCH v5] Add protections in xlog record APIs against large numbers
 and overflows.

Before this; it was possible for an extension to create malicious WAL records
that were too large to replay; or that would overflow the xl_tot_len field,
causing potential corruption in WAL record IO ops. Emitting invalid records
was also possible through pg_logical_emit_message(), which allowed you to emit
arbitrary amounts of data up to 2GB, much higher than the replay limit of 1GB.
---
 src/backend/access/transam/xloginsert.c | 63 +++++++++++++++++++++----
 src/backend/access/transam/xlogreader.c | 12 ++++-
 src/include/access/xloginsert.h         |  4 +-
 src/include/access/xlogreader.h         | 36 +++++++++++++-
 src/include/access/xlogrecord.h         |  4 ++
 5 files changed, 105 insertions(+), 14 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 2ce9be2cc7..f8d7bf9671 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -141,7 +141,18 @@ static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info,
 									   bool *topxid_included);
 static bool XLogCompressBackupBlock(char *page, uint16 hole_offset,
 									uint16 hole_length, char *dest, uint16 *dlen);
+static inline void XLogErrorDataLimitExceeded();
 
+/*
+ * Error due to exceeding the maximum size of a WAL record, or registering
+ * more datas than are being accounted for by the XLog infrastructure.
+ */
+static inline void
+XLogErrorDataLimitExceeded()
+{
+	elog(ERROR, "too much WAL data");
+}
+ 
 /*
  * Begin constructing a WAL record. This must be called before the
  * XLogRegister* functions and XLogInsert().
@@ -348,14 +359,25 @@ XLogRegisterBlock(uint8 block_id, RelFileNode *rnode, ForkNumber forknum,
  * XLogRecGetData().
  */
 void
-XLogRegisterData(char *data, int len)
+XLogRegisterData(char *data, uint32 len)
 {
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && XLogRecLengthIsValid(len));
+
+	/*
+	 * Check against max_rdatas; and ensure we don't fill a record with
+	 * more data than can be replayed. Records are allocated in one chunk
+	 * with some overhead, so ensure XLogRecLengthIsValid() for that size
+	 * of record.
+	 *
+	 * We also protect against overflows by first casting to uint64 - with only
+	 * uint32-operations this could still be an issue.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(!XLogRecLengthIsValid((uint64) mainrdata_len + (uint64) len)))
+		XLogErrorDataLimitExceeded();
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -386,12 +408,12 @@ XLogRegisterData(char *data, int len)
  * limited)
  */
 void
-XLogRegisterBufData(uint8 block_id, char *data, int len)
+XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 {
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len <= UINT16_MAX);
 
 	/* find the registered buffer struct */
 	regbuf = &registered_buffers[block_id];
@@ -399,8 +421,16 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 		elog(ERROR, "no block with id %d registered with WAL insertion",
 			 block_id);
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
+	/*
+	 * Check against max_rdatas; and ensure we don't register more data per
+	 * buffer than can be handled by the physical data format; 
+	 * i.e. that regbuf->rdata_len does not grow beyond what
+	 * XLogRecordBlockHeader->data_length can hold.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(regbuf->rdata_len + len > UINT16_MAX))
+		XLogErrorDataLimitExceeded();
+
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -519,7 +549,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, bool *topxid_included)
 {
 	XLogRecData *rdt;
-	uint32		total_len = 0;
+	uint64		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
@@ -756,12 +786,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 
 		if (needs_data)
 		{
+			/*
+			 * When copying to XLogRecordBlockHeader, the length is narrowed
+			 * to an uint16. We double-check that that is still correct.
+			 */
+			Assert(regbuf->rdata_len <= UINT16_MAX);
+
 			/*
 			 * Link the caller-supplied rdata chain for this buffer to the
 			 * overall list.
 			 */
 			bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
-			bkpb.data_length = regbuf->rdata_len;
+			bkpb.data_length = (uint16) regbuf->rdata_len;
 			total_len += regbuf->rdata_len;
 
 			rdt_datas_last->next = regbuf->rdata_head;
@@ -858,6 +894,13 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that xlogreader.c can read the record by ensuring that the
+	 * data section of the WAL record (plus infra overhead) can be allocated.
+	 */
+	if (unlikely(!XLogRecLengthIsValid(total_len)))
+		XLogErrorDataLimitExceeded();
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index cf5db23cb8..63a4ff0a0d 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -669,7 +669,17 @@ restart:
 			report_invalid_record(state,
 								  "invalid record length at %X/%X: wanted %u, got %u",
 								  LSN_FORMAT_ARGS(RecPtr),
-								  (uint32) SizeOfXLogRecord, total_len);
+								  (uint32) SizeOfXLogRecord,
+								  total_len);
+			goto err;
+		}
+		if (!XLogRecLengthIsValid(total_len))
+		{
+			report_invalid_record(state,
+								  "invalid record length at %X/%X: wanted %u, got %u",
+								  LSN_FORMAT_ARGS(RecPtr),
+								  (uint32) XLogRecMaxLength,
+								  total_len);
 			goto err;
 		}
 		gotheader = false;
diff --git a/src/include/access/xloginsert.h b/src/include/access/xloginsert.h
index 5fc340c434..5ca0b0c2f0 100644
--- a/src/include/access/xloginsert.h
+++ b/src/include/access/xloginsert.h
@@ -43,12 +43,12 @@ extern void XLogBeginInsert(void);
 extern void XLogSetRecordFlags(uint8 flags);
 extern XLogRecPtr XLogInsert(RmgrId rmid, uint8 info);
 extern void XLogEnsureRecordSpace(int max_block_id, int ndatas);
-extern void XLogRegisterData(char *data, int len);
+extern void XLogRegisterData(char *data, uint32 len);
 extern void XLogRegisterBuffer(uint8 block_id, Buffer buffer, uint8 flags);
 extern void XLogRegisterBlock(uint8 block_id, RelFileNode *rnode,
 							  ForkNumber forknum, BlockNumber blknum, char *page,
 							  uint8 flags);
-extern void XLogRegisterBufData(uint8 block_id, char *data, int len);
+extern void XLogRegisterBufData(uint8 block_id, char *data, uint32 len);
 extern void XLogResetInsertion(void);
 extern bool XLogCheckBufferNeedsBackup(Buffer buffer);
 
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index e73ea4a840..b2939b13c5 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -36,6 +36,7 @@
 
 #ifndef FRONTEND
 #include "access/transam.h"
+#include "utils/memutils.h"
 #endif
 
 #include "access/xlogrecord.h"
@@ -425,7 +426,40 @@ extern bool DecodeXLogRecord(XLogReaderState *state,
 
 #ifndef FRONTEND
 extern FullTransactionId XLogRecGetFullXid(XLogReaderState *record);
-#endif
+#else
+#define MaxAllocSize (Size) 0x3FFFFFFF
+#endif /* ifndef FRONTEND */
+
+/*
+ * XLogRecMaxSize - the maximum WAL record size that the xlogreader
+ * infrastructure can handle.
+ *
+ * Defined by MaxAllocSize and the various overheads used in the
+ * DecodedXLogRecord infrastructure.
+ *
+ * see DecodeXLogRecordRequiredSpace and allocate_recordbuf for places where
+ * allocations of full records + overhead are done.
+ */
+#define XLogRecMaxLength \
+( \
+	Min( \
+		(MaxAllocSize - ( \
+			offsetof(DecodedXLogRecord, blocks[0]) + \
+			(sizeof(DecodedBkpBlock) * (XLR_MAX_BLOCK_ID + 1)) + \
+			(MAXIMUM_ALIGNOF - 1) + \
+			(MAXIMUM_ALIGNOF - 1) + \
+			(MAXIMUM_ALIGNOF - 1) \
+		)), \
+		MaxAllocSize - (MaxAllocSize % XLOG_BLCKSZ) \
+	) \
+)
+
+/*
+ * XLogRecLengthIsValid
+ *
+ * Check that this length does not exceed the maximum xlog record length
+ */
+#define XLogRecLengthIsValid(xlr_size) ((xlr_size) <= XLogRecMaxLength)
 
 extern bool RestoreBlockImage(XLogReaderState *record, uint8 block_id, char *page);
 extern char *XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len);
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 052ac6817a..3575363e84 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -37,6 +37,10 @@
  * The XLogRecordBlockHeader, XLogRecordDataHeaderShort and
  * XLogRecordDataHeaderLong structs all begin with a single 'id' byte. It's
  * used to distinguish between block references, and the main data structs.
+ *
+ * Note that xlogreader.c limits the total size of the xl_tot_len to
+ * MaxAllocSize (1GB - 1). This means that this is also the maximum size
+ * of a single WAL record - we would be unable to replay any larger record.
  */
 typedef struct XLogRecord
 {
-- 
2.30.2

From 321cdb85893b57e959f8633d9ffd866f8be2b3fd Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Mon, 20 Jun 2022 10:21:22 +0200
Subject: [PATCH v5] Add protections in xlog record APIs against large numbers
 and overflows.

Before this; it was possible for an extension to create malicious WAL records
that were too large to replay; or that would overflow the xl_tot_len field,
causing potential corruption in WAL record IO ops. Emitting invalid records
was also possible through pg_logical_emit_message(), which allowed you to emit
arbitrary amounts of data up to 2GB, much higher than the replay limit of 1GB.
---
 src/backend/access/transam/xloginsert.c | 59 +++++++++++++++++++++----
 src/backend/access/transam/xlogreader.c | 12 ++++-
 src/include/access/xlogreader.h         | 36 ++++++++++++++-
 src/include/access/xlogrecord.h         |  4 ++
 4 files changed, 101 insertions(+), 10 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 2ce9be2cc7..a8ac036cbd 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -141,7 +141,18 @@ static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info,
 									   bool *topxid_included);
 static bool XLogCompressBackupBlock(char *page, uint16 hole_offset,
 									uint16 hole_length, char *dest, uint16 *dlen);
+static inline void XLogErrorDataLimitExceeded();
 
+/*
+ * Error due to exceeding the maximum size of a WAL record, or registering
+ * more datas than are being accounted for by the XLog infrastructure.
+ */
+static inline void
+XLogErrorDataLimitExceeded()
+{
+	elog(ERROR, "too much WAL data");
+}
+ 
 /*
  * Begin constructing a WAL record. This must be called before the
  * XLogRegister* functions and XLogInsert().
@@ -352,10 +363,21 @@ XLogRegisterData(char *data, int len)
 {
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len >= 0 && XLogRecLengthIsValid(len));
+
+	/*
+	 * Check against max_rdatas; and ensure we don't fill a record with
+	 * more data than can be replayed. Records are allocated in one chunk
+	 * with some overhead, so ensure XLogRecLengthIsValid() for that size
+	 * of record.
+	 *
+	 * We also protect against overflows by first casting to uint64 - with only
+	 * uint32-operations this could still be an issue.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(!XLogRecLengthIsValid((uint64) mainrdata_len + (uint64) len)))
+		XLogErrorDataLimitExceeded();
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -391,7 +413,7 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len >= 0 && len <= UINT16_MAX);
 
 	/* find the registered buffer struct */
 	regbuf = &registered_buffers[block_id];
@@ -399,8 +421,16 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 		elog(ERROR, "no block with id %d registered with WAL insertion",
 			 block_id);
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
+	/*
+	 * Check against max_rdatas; and ensure we don't register more data per
+	 * buffer than can be handled by the physical data format; 
+	 * i.e. that regbuf->rdata_len does not grow beyond what
+	 * XLogRecordBlockHeader->data_length can hold.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(regbuf->rdata_len + len > UINT16_MAX))
+		XLogErrorDataLimitExceeded();
+
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -519,7 +549,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, bool *topxid_included)
 {
 	XLogRecData *rdt;
-	uint32		total_len = 0;
+	uint64		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
@@ -756,12 +786,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 
 		if (needs_data)
 		{
+			/*
+			 * When copying to XLogRecordBlockHeader, the length is narrowed
+			 * to an uint16. We double-check that that is still correct.
+			 */
+			Assert(regbuf->rdata_len <= UINT16_MAX);
+
 			/*
 			 * Link the caller-supplied rdata chain for this buffer to the
 			 * overall list.
 			 */
 			bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
-			bkpb.data_length = regbuf->rdata_len;
+			bkpb.data_length = (uint16) regbuf->rdata_len;
 			total_len += regbuf->rdata_len;
 
 			rdt_datas_last->next = regbuf->rdata_head;
@@ -858,6 +894,13 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that xlogreader.c can read the record by ensuring that the
+	 * data section of the WAL record (plus infra overhead) can be allocated.
+	 */
+	if (unlikely(!XLogRecLengthIsValid(total_len)))
+		XLogErrorDataLimitExceeded();
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index cf5db23cb8..63a4ff0a0d 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -669,7 +669,17 @@ restart:
 			report_invalid_record(state,
 								  "invalid record length at %X/%X: wanted %u, got %u",
 								  LSN_FORMAT_ARGS(RecPtr),
-								  (uint32) SizeOfXLogRecord, total_len);
+								  (uint32) SizeOfXLogRecord,
+								  total_len);
+			goto err;
+		}
+		if (!XLogRecLengthIsValid(total_len))
+		{
+			report_invalid_record(state,
+								  "invalid record length at %X/%X: wanted %u, got %u",
+								  LSN_FORMAT_ARGS(RecPtr),
+								  (uint32) XLogRecMaxLength,
+								  total_len);
 			goto err;
 		}
 		gotheader = false;
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index e73ea4a840..b2939b13c5 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -36,6 +36,7 @@
 
 #ifndef FRONTEND
 #include "access/transam.h"
+#include "utils/memutils.h"
 #endif
 
 #include "access/xlogrecord.h"
@@ -425,7 +426,40 @@ extern bool DecodeXLogRecord(XLogReaderState *state,
 
 #ifndef FRONTEND
 extern FullTransactionId XLogRecGetFullXid(XLogReaderState *record);
-#endif
+#else
+#define MaxAllocSize (Size) 0x3FFFFFFF
+#endif /* ifndef FRONTEND */
+
+/*
+ * XLogRecMaxSize - the maximum WAL record size that the xlogreader
+ * infrastructure can handle.
+ *
+ * Defined by MaxAllocSize and the various overheads used in the
+ * DecodedXLogRecord infrastructure.
+ *
+ * see DecodeXLogRecordRequiredSpace and allocate_recordbuf for places where
+ * allocations of full records + overhead are done.
+ */
+#define XLogRecMaxLength \
+( \
+	Min( \
+		(MaxAllocSize - ( \
+			offsetof(DecodedXLogRecord, blocks[0]) + \
+			(sizeof(DecodedBkpBlock) * (XLR_MAX_BLOCK_ID + 1)) + \
+			(MAXIMUM_ALIGNOF - 1) + \
+			(MAXIMUM_ALIGNOF - 1) + \
+			(MAXIMUM_ALIGNOF - 1) \
+		)), \
+		MaxAllocSize - (MaxAllocSize % XLOG_BLCKSZ) \
+	) \
+)
+
+/*
+ * XLogRecLengthIsValid
+ *
+ * Check that this length does not exceed the maximum xlog record length
+ */
+#define XLogRecLengthIsValid(xlr_size) ((xlr_size) <= XLogRecMaxLength)
 
 extern bool RestoreBlockImage(XLogReaderState *record, uint8 block_id, char *page);
 extern char *XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len);
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 052ac6817a..3575363e84 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -37,6 +37,10 @@
  * The XLogRecordBlockHeader, XLogRecordDataHeaderShort and
  * XLogRecordDataHeaderLong structs all begin with a single 'id' byte. It's
  * used to distinguish between block references, and the main data structs.
+ *
+ * Note that xlogreader.c limits the total size of the xl_tot_len to
+ * MaxAllocSize (1GB - 1). This means that this is also the maximum size
+ * of a single WAL record - we would be unable to replay any larger record.
  */
 typedef struct XLogRecord
 {
-- 
2.30.2

From af666e94f54b6bb81f1a6a86c39c4cfb2db9e799 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Mon, 20 Jun 2022 10:21:22 +0200
Subject: [PATCH v6] Add protections in xlog record APIs against large numbers
 and overflows.

Before this; it was possible for an extension to create malicious WAL records
that were too large to replay; or that would overflow the xl_tot_len field,
causing potential corruption in WAL record IO ops. Emitting invalid records
was also possible through pg_logical_emit_message(), which allowed you to emit
arbitrary amounts of data up to 2GB, much higher than the replay limit of 1GB.
---
 src/backend/access/transam/xloginsert.c | 63 +++++++++++++++++++++----
 src/backend/access/transam/xlogreader.c | 12 ++++-
 src/include/access/xloginsert.h         |  4 +-
 src/include/access/xlogreader.h         | 36 +++++++++++++-
 src/include/access/xlogrecord.h         |  4 ++
 5 files changed, 105 insertions(+), 14 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index f3c29fa909..814d83b362 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -141,7 +141,18 @@ static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info,
 									   bool *topxid_included);
 static bool XLogCompressBackupBlock(char *page, uint16 hole_offset,
 									uint16 hole_length, char *dest, uint16 *dlen);
+static inline void XLogErrorDataLimitExceeded();
 
+/*
+ * Error due to exceeding the maximum size of a WAL record, or registering
+ * more datas than are being accounted for by the XLog infrastructure.
+ */
+static inline void
+XLogErrorDataLimitExceeded()
+{
+	elog(ERROR, "too much WAL data");
+}
+ 
 /*
  * Begin constructing a WAL record. This must be called before the
  * XLogRegister* functions and XLogInsert().
@@ -348,14 +359,25 @@ XLogRegisterBlock(uint8 block_id, RelFileLocator *rlocator, ForkNumber forknum,
  * XLogRecGetData().
  */
 void
-XLogRegisterData(char *data, int len)
+XLogRegisterData(char *data, uint32 len)
 {
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && XLogRecLengthIsValid(len));
+
+	/*
+	 * Check against max_rdatas; and ensure we don't fill a record with
+	 * more data than can be replayed. Records are allocated in one chunk
+	 * with some overhead, so ensure XLogRecLengthIsValid() for that size
+	 * of record.
+	 *
+	 * We also protect against overflows by first casting to uint64 - with only
+	 * uint32-operations this could still be an issue.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(!XLogRecLengthIsValid((uint64) mainrdata_len + (uint64) len)))
+		XLogErrorDataLimitExceeded();
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -386,12 +408,12 @@ XLogRegisterData(char *data, int len)
  * limited)
  */
 void
-XLogRegisterBufData(uint8 block_id, char *data, int len)
+XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 {
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len <= UINT16_MAX);
 
 	/* find the registered buffer struct */
 	regbuf = &registered_buffers[block_id];
@@ -399,8 +421,16 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 		elog(ERROR, "no block with id %d registered with WAL insertion",
 			 block_id);
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
+	/*
+	 * Check against max_rdatas; and ensure we don't register more data per
+	 * buffer than can be handled by the physical data format; 
+	 * i.e. that regbuf->rdata_len does not grow beyond what
+	 * XLogRecordBlockHeader->data_length can hold.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(regbuf->rdata_len + len > UINT16_MAX))
+		XLogErrorDataLimitExceeded();
+
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -519,7 +549,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, bool *topxid_included)
 {
 	XLogRecData *rdt;
-	uint32		total_len = 0;
+	uint64		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
@@ -756,12 +786,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 
 		if (needs_data)
 		{
+			/*
+			 * When copying to XLogRecordBlockHeader, the length is narrowed
+			 * to an uint16. We double-check that that is still correct.
+			 */
+			Assert(regbuf->rdata_len <= UINT16_MAX);
+
 			/*
 			 * Link the caller-supplied rdata chain for this buffer to the
 			 * overall list.
 			 */
 			bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
-			bkpb.data_length = regbuf->rdata_len;
+			bkpb.data_length = (uint16) regbuf->rdata_len;
 			total_len += regbuf->rdata_len;
 
 			rdt_datas_last->next = regbuf->rdata_head;
@@ -858,6 +894,13 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that xlogreader.c can read the record by ensuring that the
+	 * data section of the WAL record (plus infra overhead) can be allocated.
+	 */
+	if (unlikely(!XLogRecLengthIsValid(total_len)))
+		XLogErrorDataLimitExceeded();
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index f3dc4b7797..deb1dff3d0 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -669,7 +669,17 @@ restart:
 			report_invalid_record(state,
 								  "invalid record length at %X/%X: wanted %u, got %u",
 								  LSN_FORMAT_ARGS(RecPtr),
-								  (uint32) SizeOfXLogRecord, total_len);
+								  (uint32) SizeOfXLogRecord,
+								  total_len);
+			goto err;
+		}
+		if (!XLogRecLengthIsValid(total_len))
+		{
+			report_invalid_record(state,
+								  "invalid record length at %X/%X: wanted %u, got %u",
+								  LSN_FORMAT_ARGS(RecPtr),
+								  (uint32) XLogRecMaxLength,
+								  total_len);
 			goto err;
 		}
 		gotheader = false;
diff --git a/src/include/access/xloginsert.h b/src/include/access/xloginsert.h
index c04f77b173..aed4643d1c 100644
--- a/src/include/access/xloginsert.h
+++ b/src/include/access/xloginsert.h
@@ -43,12 +43,12 @@ extern void XLogBeginInsert(void);
 extern void XLogSetRecordFlags(uint8 flags);
 extern XLogRecPtr XLogInsert(RmgrId rmid, uint8 info);
 extern void XLogEnsureRecordSpace(int max_block_id, int ndatas);
-extern void XLogRegisterData(char *data, int len);
+extern void XLogRegisterData(char *data, uint32 len);
 extern void XLogRegisterBuffer(uint8 block_id, Buffer buffer, uint8 flags);
 extern void XLogRegisterBlock(uint8 block_id, RelFileLocator *rlocator,
 							  ForkNumber forknum, BlockNumber blknum, char *page,
 							  uint8 flags);
-extern void XLogRegisterBufData(uint8 block_id, char *data, int len);
+extern void XLogRegisterBufData(uint8 block_id, char *data, uint32 len);
 extern void XLogResetInsertion(void);
 extern bool XLogCheckBufferNeedsBackup(Buffer buffer);
 
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index 5395f155aa..8f7ce4e5fe 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -36,6 +36,7 @@
 
 #ifndef FRONTEND
 #include "access/transam.h"
+#include "utils/memutils.h"
 #endif
 
 #include "access/xlogrecord.h"
@@ -425,7 +426,40 @@ extern bool DecodeXLogRecord(XLogReaderState *state,
 
 #ifndef FRONTEND
 extern FullTransactionId XLogRecGetFullXid(XLogReaderState *record);
-#endif
+#else
+#define MaxAllocSize (Size) 0x3FFFFFFF
+#endif /* ifndef FRONTEND */
+
+/*
+ * XLogRecMaxSize - the maximum WAL record size that the xlogreader
+ * infrastructure can handle.
+ *
+ * Defined by MaxAllocSize and the various overheads used in the
+ * DecodedXLogRecord infrastructure.
+ *
+ * see DecodeXLogRecordRequiredSpace and allocate_recordbuf for places where
+ * allocations of full records + overhead are done.
+ */
+#define XLogRecMaxLength \
+( \
+	Min( \
+		(MaxAllocSize - ( \
+			offsetof(DecodedXLogRecord, blocks[0]) + \
+			(sizeof(DecodedBkpBlock) * (XLR_MAX_BLOCK_ID + 1)) + \
+			(MAXIMUM_ALIGNOF - 1) + \
+			(MAXIMUM_ALIGNOF - 1) + \
+			(MAXIMUM_ALIGNOF - 1) \
+		)), \
+		MaxAllocSize - (MaxAllocSize % XLOG_BLCKSZ) \
+	) \
+)
+
+/*
+ * XLogRecLengthIsValid
+ *
+ * Check that this length does not exceed the maximum xlog record length
+ */
+#define XLogRecLengthIsValid(xlr_size) ((xlr_size) <= XLogRecMaxLength)
 
 extern bool RestoreBlockImage(XLogReaderState *record, uint8 block_id, char *page);
 extern char *XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len);
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 835151ec92..2a282dac0c 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -37,6 +37,10 @@
  * The XLogRecordBlockHeader, XLogRecordDataHeaderShort and
  * XLogRecordDataHeaderLong structs all begin with a single 'id' byte. It's
  * used to distinguish between block references, and the main data structs.
+ *
+ * Note that xlogreader.c limits the total size of the xl_tot_len to
+ * MaxAllocSize (1GB - 1). This means that this is also the maximum size
+ * of a single WAL record - we would be unable to replay any larger record.
  */
 typedef struct XLogRecord
 {
-- 
2.30.2

From fa9a18dfa478f88b66e3f831057ecd9e986147b3 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Mon, 20 Jun 2022 10:21:22 +0200
Subject: [PATCH v6] Add protections in xlog record APIs against large numbers
 and overflows.

Before this; it was possible for an extension to create malicious WAL records
that were too large to replay; or that would overflow the xl_tot_len field,
causing potential corruption in WAL record IO ops. Emitting invalid records
was also possible through pg_logical_emit_message(), which allowed you to emit
arbitrary amounts of data up to 2GB, much higher than the replay limit of 1GB.
---
 src/backend/access/transam/xloginsert.c | 59 +++++++++++++++++++++----
 src/backend/access/transam/xlogreader.c | 12 ++++-
 src/include/access/xlogreader.h         | 36 ++++++++++++++-
 src/include/access/xlogrecord.h         |  4 ++
 4 files changed, 101 insertions(+), 10 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 2ce9be2cc7..a8ac036cbd 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -141,7 +141,18 @@ static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info,
 									   bool *topxid_included);
 static bool XLogCompressBackupBlock(char *page, uint16 hole_offset,
 									uint16 hole_length, char *dest, uint16 *dlen);
+static inline void XLogErrorDataLimitExceeded();
 
+/*
+ * Error due to exceeding the maximum size of a WAL record, or registering
+ * more datas than are being accounted for by the XLog infrastructure.
+ */
+static inline void
+XLogErrorDataLimitExceeded()
+{
+	elog(ERROR, "too much WAL data");
+}
+ 
 /*
  * Begin constructing a WAL record. This must be called before the
  * XLogRegister* functions and XLogInsert().
@@ -352,10 +363,21 @@ XLogRegisterData(char *data, int len)
 {
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len >= 0 && XLogRecLengthIsValid(len));
+
+	/*
+	 * Check against max_rdatas; and ensure we don't fill a record with
+	 * more data than can be replayed. Records are allocated in one chunk
+	 * with some overhead, so ensure XLogRecLengthIsValid() for that size
+	 * of record.
+	 *
+	 * We also protect against overflows by first casting to uint64 - with only
+	 * uint32-operations this could still be an issue.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(!XLogRecLengthIsValid((uint64) mainrdata_len + (uint64) len)))
+		XLogErrorDataLimitExceeded();
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -391,7 +413,7 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len >= 0 && len <= UINT16_MAX);
 
 	/* find the registered buffer struct */
 	regbuf = &registered_buffers[block_id];
@@ -399,8 +421,16 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 		elog(ERROR, "no block with id %d registered with WAL insertion",
 			 block_id);
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
+	/*
+	 * Check against max_rdatas; and ensure we don't register more data per
+	 * buffer than can be handled by the physical data format; 
+	 * i.e. that regbuf->rdata_len does not grow beyond what
+	 * XLogRecordBlockHeader->data_length can hold.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(regbuf->rdata_len + len > UINT16_MAX))
+		XLogErrorDataLimitExceeded();
+
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -519,7 +549,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, bool *topxid_included)
 {
 	XLogRecData *rdt;
-	uint32		total_len = 0;
+	uint64		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
@@ -756,12 +786,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 
 		if (needs_data)
 		{
+			/*
+			 * When copying to XLogRecordBlockHeader, the length is narrowed
+			 * to an uint16. We double-check that that is still correct.
+			 */
+			Assert(regbuf->rdata_len <= UINT16_MAX);
+
 			/*
 			 * Link the caller-supplied rdata chain for this buffer to the
 			 * overall list.
 			 */
 			bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
-			bkpb.data_length = regbuf->rdata_len;
+			bkpb.data_length = (uint16) regbuf->rdata_len;
 			total_len += regbuf->rdata_len;
 
 			rdt_datas_last->next = regbuf->rdata_head;
@@ -858,6 +894,13 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that xlogreader.c can read the record by ensuring that the
+	 * data section of the WAL record (plus infra overhead) can be allocated.
+	 */
+	if (unlikely(!XLogRecLengthIsValid(total_len)))
+		XLogErrorDataLimitExceeded();
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index cf5db23cb8..63a4ff0a0d 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -669,7 +669,17 @@ restart:
 			report_invalid_record(state,
 								  "invalid record length at %X/%X: wanted %u, got %u",
 								  LSN_FORMAT_ARGS(RecPtr),
-								  (uint32) SizeOfXLogRecord, total_len);
+								  (uint32) SizeOfXLogRecord,
+								  total_len);
+			goto err;
+		}
+		if (!XLogRecLengthIsValid(total_len))
+		{
+			report_invalid_record(state,
+								  "invalid record length at %X/%X: wanted %u, got %u",
+								  LSN_FORMAT_ARGS(RecPtr),
+								  (uint32) XLogRecMaxLength,
+								  total_len);
 			goto err;
 		}
 		gotheader = false;
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index e73ea4a840..b2939b13c5 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -36,6 +36,7 @@
 
 #ifndef FRONTEND
 #include "access/transam.h"
+#include "utils/memutils.h"
 #endif
 
 #include "access/xlogrecord.h"
@@ -425,7 +426,40 @@ extern bool DecodeXLogRecord(XLogReaderState *state,
 
 #ifndef FRONTEND
 extern FullTransactionId XLogRecGetFullXid(XLogReaderState *record);
-#endif
+#else
+#define MaxAllocSize (Size) 0x3FFFFFFF
+#endif /* ifndef FRONTEND */
+
+/*
+ * XLogRecMaxSize - the maximum WAL record size that the xlogreader
+ * infrastructure can handle.
+ *
+ * Defined by MaxAllocSize and the various overheads used in the
+ * DecodedXLogRecord infrastructure.
+ *
+ * see DecodeXLogRecordRequiredSpace and allocate_recordbuf for places where
+ * allocations of full records + overhead are done.
+ */
+#define XLogRecMaxLength \
+( \
+	Min( \
+		(MaxAllocSize - ( \
+			offsetof(DecodedXLogRecord, blocks[0]) + \
+			(sizeof(DecodedBkpBlock) * (XLR_MAX_BLOCK_ID + 1)) + \
+			(MAXIMUM_ALIGNOF - 1) + \
+			(MAXIMUM_ALIGNOF - 1) + \
+			(MAXIMUM_ALIGNOF - 1) \
+		)), \
+		MaxAllocSize - (MaxAllocSize % XLOG_BLCKSZ) \
+	) \
+)
+
+/*
+ * XLogRecLengthIsValid
+ *
+ * Check that this length does not exceed the maximum xlog record length
+ */
+#define XLogRecLengthIsValid(xlr_size) ((xlr_size) <= XLogRecMaxLength)
 
 extern bool RestoreBlockImage(XLogReaderState *record, uint8 block_id, char *page);
 extern char *XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len);
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 052ac6817a..3575363e84 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -37,6 +37,10 @@
  * The XLogRecordBlockHeader, XLogRecordDataHeaderShort and
  * XLogRecordDataHeaderLong structs all begin with a single 'id' byte. It's
  * used to distinguish between block references, and the main data structs.
+ *
+ * Note that xlogreader.c limits the total size of the xl_tot_len to
+ * MaxAllocSize (1GB - 1). This means that this is also the maximum size
+ * of a single WAL record - we would be unable to replay any larger record.
  */
 typedef struct XLogRecord
 {
-- 
2.30.2

From 21a933ab3d2c0d9cc2b7c06329df28145f5930df Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Mon, 20 Jun 2022 10:21:22 +0200
Subject: [PATCH v6] Add protections in xlog record APIs against large numbers
 and overflows.

Before this; it was possible for an extension to create malicious WAL records
that were too large to replay; or that would overflow the xl_tot_len field,
causing potential corruption in WAL record IO ops. Emitting invalid records
was also possible through pg_logical_emit_message(), which allowed you to emit
arbitrary amounts of data up to 2GB, much higher than the replay limit of 1GB.
---
 src/backend/access/transam/xloginsert.c | 59 +++++++++++++++++++++----
 src/backend/access/transam/xlogreader.c | 12 ++++-
 src/include/access/xlogreader.h         | 31 ++++++++++++-
 src/include/access/xlogrecord.h         |  4 ++
 4 files changed, 96 insertions(+), 10 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index b153fad594..f73785911a 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -114,7 +114,18 @@ static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info,
 									   XLogRecPtr *fpw_lsn, int *num_fpi);
 static bool XLogCompressBackupBlock(char *page, uint16 hole_offset,
 									uint16 hole_length, char *dest, uint16 *dlen);
+static inline void XLogErrorDataLimitExceeded();
 
+/*
+ * Error due to exceeding the maximum size of a WAL record, or registering
+ * more datas than are being accounted for by the XLog infrastructure.
+ */
+static inline void
+XLogErrorDataLimitExceeded()
+{
+	elog(ERROR, "too much WAL data");
+}
+ 
 /*
  * Begin constructing a WAL record. This must be called before the
  * XLogRegister* functions and XLogInsert().
@@ -331,10 +342,21 @@ XLogRegisterData(char *data, int len)
 {
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len >= 0 && XLogRecLengthIsValid(len));
+
+	/*
+	 * Check against max_rdatas; and ensure we don't fill a record with
+	 * more data than can be replayed. Records are allocated in one chunk
+	 * with some overhead, so ensure XLogRecLengthIsValid() for that size
+	 * of record.
+	 *
+	 * We also protect against overflows by first casting to uint64 - with only
+	 * uint32-operations this could still be an issue.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(!XLogRecLengthIsValid((uint64) mainrdata_len + (uint64) len)))
+		XLogErrorDataLimitExceeded();
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -370,7 +392,7 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len >= 0 && len <= UINT16_MAX);
 
 	/* find the registered buffer struct */
 	regbuf = &registered_buffers[block_id];
@@ -378,8 +400,16 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 		elog(ERROR, "no block with id %d registered with WAL insertion",
 			 block_id);
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
+	/*
+	 * Check against max_rdatas; and ensure we don't register more data per
+	 * buffer than can be handled by the physical data format; 
+	 * i.e. that regbuf->rdata_len does not grow beyond what
+	 * XLogRecordBlockHeader->data_length can hold.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(regbuf->rdata_len + len > UINT16_MAX))
+		XLogErrorDataLimitExceeded();
+
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -495,7 +525,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi)
 {
 	XLogRecData *rdt;
-	uint32		total_len = 0;
+	uint64		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
@@ -703,12 +733,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 
 		if (needs_data)
 		{
+			/*
+			 * When copying to XLogRecordBlockHeader, the length is narrowed
+			 * to an uint16. We double-check that that is still correct.
+			 */
+			Assert(regbuf->rdata_len <= UINT16_MAX);
+
 			/*
 			 * Link the caller-supplied rdata chain for this buffer to the
 			 * overall list.
 			 */
 			bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
-			bkpb.data_length = regbuf->rdata_len;
+			bkpb.data_length = (uint16) regbuf->rdata_len;
 			total_len += regbuf->rdata_len;
 
 			rdt_datas_last->next = regbuf->rdata_head;
@@ -805,6 +841,13 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that xlogreader.c can read the record by ensuring that the
+	 * data section of the WAL record (plus infra overhead) can be allocated.
+	 */
+	if (unlikely(!XLogRecLengthIsValid(total_len)))
+		XLogErrorDataLimitExceeded();
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
index d797d9d508..1e39b3033e 100644
--- a/src/backend/access/transam/xlogreader.c
+++ b/src/backend/access/transam/xlogreader.c
@@ -402,7 +402,17 @@ restart:
 			report_invalid_record(state,
 								  "invalid record length at %X/%X: wanted %u, got %u",
 								  LSN_FORMAT_ARGS(RecPtr),
-								  (uint32) SizeOfXLogRecord, total_len);
+								  (uint32) SizeOfXLogRecord,
+								  total_len);
+			goto err;
+		}
+		if (!XLogRecLengthIsValid(total_len))
+		{
+			report_invalid_record(state,
+								  "invalid record length at %X/%X: wanted %u, got %u",
+								  LSN_FORMAT_ARGS(RecPtr),
+								  (uint32) XLogRecMaxLength,
+								  total_len);
 			goto err;
 		}
 		gotheader = false;
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h
index 10458c23ed..1cbc70a04e 100644
--- a/src/include/access/xlogreader.h
+++ b/src/include/access/xlogreader.h
@@ -36,6 +36,7 @@
 
 #ifndef FRONTEND
 #include "access/transam.h"
+#include "utils/memutils.h"
 #endif
 
 #include "access/xlogrecord.h"
@@ -329,7 +330,35 @@ extern bool DecodeXLogRecord(XLogReaderState *state, XLogRecord *record,
 
 #ifndef FRONTEND
 extern FullTransactionId XLogRecGetFullXid(XLogReaderState *record);
-#endif
+#else
+#define MaxAllocSize (Size) 0x3FFFFFFF
+#endif /* ifndef FRONTEND */
+
+/*
+ * XLogRecMaxSize - the maximum WAL record size that the xlogreader
+ * infrastructure can handle.
+ *
+ * Defined by MaxAllocSize and the various overheads used in the
+ * DecodedXLogRecord infrastructure.
+ *
+ * Places that allocate memory based on xlogrecords' data lengths:
+ *	 - allocate_recordbuf 
+ *		Allocates xl_tot_len, rounded up to the next multiple of
+ *		XLOG_BLCKSZ.
+ *	 - DecodeXLogRecord
+ *		Allocates a buffer for xlog main data only, rounded up
+ *		to the next multiple of BLCKSZ/2.
+ */
+#define XLogRecMaxLength \
+	(MaxAllocSize - (MaxAllocSize % XLOG_BLCKSZ))
+
+
+/*
+ * XLogRecLengthIsValid
+ *
+ * Check that this length does not exceed the maximum xlog record length
+ */
+#define XLogRecLengthIsValid(xlr_size) ((xlr_size) <= XLogRecMaxLength)
 
 extern bool RestoreBlockImage(XLogReaderState *record, uint8 block_id, char *page);
 extern char *XLogRecGetBlockData(XLogReaderState *record, uint8 block_id, Size *len);
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index f68cb18912..2f60a3ab39 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -37,6 +37,10 @@
  * The XLogRecordBlockHeader, XLogRecordDataHeaderShort and
  * XLogRecordDataHeaderLong structs all begin with a single 'id' byte. It's
  * used to distinguish between block references, and the main data structs.
+ *
+ * Note that xlogreader.c limits the total size of the xl_tot_len to
+ * MaxAllocSize (1GB - 1). This means that this is also the maximum size
+ * of a single WAL record - we would be unable to replay any larger record.
  */
 typedef struct XLogRecord
 {
-- 
2.30.2

From 2031f15fe90c94deb21d4e41b717ada9a8bdb520 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Mon, 20 Jun 2022 10:21:22 +0200
Subject: [PATCH v7] Add protections in xlog record APIs against large numbers
 and overflows.

Before this; it was possible for an extension to create malicious WAL records
that were too large to replay; or that would overflow the xl_tot_len field,
causing potential corruption in WAL record IO ops. Emitting invalid records
was also possible through pg_logical_emit_message(), which allowed you to emit
arbitrary amounts of data up to 2GB, much higher than the replay limit of 1GB.
---
 src/include/access/xloginsert.h         |  4 ++--
 src/backend/access/transam/xloginsert.c | 30 ++++++++++++++++++++-----
 2 files changed, 27 insertions(+), 7 deletions(-)

diff --git a/src/include/access/xloginsert.h b/src/include/access/xloginsert.h
index c04f77b173..aed4643d1c 100644
--- a/src/include/access/xloginsert.h
+++ b/src/include/access/xloginsert.h
@@ -43,12 +43,12 @@ extern void XLogBeginInsert(void);
 extern void XLogSetRecordFlags(uint8 flags);
 extern XLogRecPtr XLogInsert(RmgrId rmid, uint8 info);
 extern void XLogEnsureRecordSpace(int max_block_id, int ndatas);
-extern void XLogRegisterData(char *data, int len);
+extern void XLogRegisterData(char *data, uint32 len);
 extern void XLogRegisterBuffer(uint8 block_id, Buffer buffer, uint8 flags);
 extern void XLogRegisterBlock(uint8 block_id, RelFileLocator *rlocator,
 							  ForkNumber forknum, BlockNumber blknum, char *page,
 							  uint8 flags);
-extern void XLogRegisterBufData(uint8 block_id, char *data, int len);
+extern void XLogRegisterBufData(uint8 block_id, char *data, uint32 len);
 extern void XLogResetInsertion(void);
 extern bool XLogCheckBufferNeedsBackup(Buffer buffer);
 
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index f3c29fa909..3788429b70 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -348,13 +348,13 @@ XLogRegisterBlock(uint8 block_id, RelFileLocator *rlocator, ForkNumber forknum,
  * XLogRecGetData().
  */
 void
-XLogRegisterData(char *data, int len)
+XLogRegisterData(char *data, uint32 len)
 {
 	XLogRecData *rdata;
 
 	Assert(begininsert_called);
 
-	if (num_rdatas >= max_rdatas)
+	if (unlikely(num_rdatas >= max_rdatas))
 		elog(ERROR, "too much WAL data");
 	rdata = &rdatas[num_rdatas++];
 
@@ -386,7 +386,7 @@ XLogRegisterData(char *data, int len)
  * limited)
  */
 void
-XLogRegisterBufData(uint8 block_id, char *data, int len)
+XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 {
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
@@ -399,8 +399,16 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 		elog(ERROR, "no block with id %d registered with WAL insertion",
 			 block_id);
 
-	if (num_rdatas >= max_rdatas)
+	/*
+	 * Check against max_rdatas and ensure we do not register more data per
+	 * buffer than can be handled by the physical data format; i.e. that
+	 * regbuf->rdata_len does not grow beyond what
+	 * XLogRecordBlockHeader->data_length can hold.
+	 */
+	if (unlikely(num_rdatas >= max_rdatas) ||
+		unlikely(regbuf->rdata_len + len > UINT16_MAX))
 		elog(ERROR, "too much WAL data");
+
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -756,12 +764,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 
 		if (needs_data)
 		{
+			/*
+			 * When copying to XLogRecordBlockHeader, the length is narrowed
+			 * to an uint16. We double-check that that is still correct.
+			 */
+			Assert(regbuf->rdata_len <= UINT16_MAX);
+
 			/*
 			 * Link the caller-supplied rdata chain for this buffer to the
 			 * overall list.
 			 */
 			bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
-			bkpb.data_length = regbuf->rdata_len;
+			bkpb.data_length = (uint16) regbuf->rdata_len;
 			total_len += regbuf->rdata_len;
 
 			rdt_datas_last->next = regbuf->rdata_head;
@@ -858,6 +872,12 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that xlogreader.c can read the record.
+	 */
+	if (unlikely(!AllocSizeIsValid(DecodeXLogRecordRequiredSpace(total_len))))
+		elog(ERROR, "too much WAL data");
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
-- 
2.36.1

From c2e24db75eb6d1f1e647173081964971a8454c9b Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Mon, 20 Jun 2022 10:21:22 +0200
Subject: [PATCH v8] Add protections in xlog record APIs against large numbers
 and overflows.

Before this, it was possible for an extension to create malicious WAL records
that were too large to replay; or that would overflow the xl_tot_len field,
causing potential corruption in WAL record IO ops.

Emitting invalid records was also possible through pg_logical_emit_message(),
which allowed you to emit arbitrary amounts of data up to 2GB, much higher
than the replay limit of approximately 1GB.

This patch adds a limit to the size of an XLogRecord (1020MiB), checks
against overflows, and ensures that the reader infrastructure can read any
max-sized XLogRecords, such that the wal-generating backend will fail when
it attempts to insert unreadable records, instead of that insertion
succeeding but breaking any replication streams.
---
 src/backend/access/transam/xloginsert.c | 77 ++++++++++++++++++++++---
 src/include/access/xloginsert.h         |  4 +-
 src/include/access/xlogrecord.h         | 13 +++++
 3 files changed, 83 insertions(+), 11 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index f3c29fa909..4b343c1c12 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -141,7 +141,18 @@ static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info,
 									   bool *topxid_included);
 static bool XLogCompressBackupBlock(char *page, uint16 hole_offset,
 									uint16 hole_length, char *dest, uint16 *dlen);
+static inline void XLogErrorDataLimitExceeded();
 
+/*
+ * Error due to exceeding the maximum size of a WAL record, or registering
+ * more datas than are being accounted for by the XLog infrastructure.
+ */
+static inline void
+XLogErrorDataLimitExceeded()
+{
+	elog(ERROR, "too much WAL data");
+}
+ 
 /*
  * Begin constructing a WAL record. This must be called before the
  * XLogRegister* functions and XLogInsert().
@@ -348,14 +359,29 @@ XLogRegisterBlock(uint8 block_id, RelFileLocator *rlocator, ForkNumber forknum,
  * XLogRecGetData().
  */
 void
-XLogRegisterData(char *data, int len)
+XLogRegisterData(char *data, uint32 len)
 {
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && XLogRecordLengthIsValid(len));
+
+	/*
+	 * Check against max_rdatas; and ensure we don't fill a record with
+	 * more data than can be replayed. Records are allocated in one chunk
+	 * with some overhead, so ensure XLogRecordLengthIsValid() for that
+	 * size of record.
+	 *
+	 * Additionally, check that we don't accidentally overflow the
+	 * intermediate sum value on 32-bit systems by ensuring that the
+	 * sum of the two inputs is no less than one of the inputs.
+	 */
+	if (num_rdatas >= max_rdatas ||
+#if SIZEOF_SIZE_T == 4
+		 mainrdata_len + len < len ||
+#endif
+		!XLogRecordLengthIsValid((size_t) mainrdata_len + (size_t) len))
+		XLogErrorDataLimitExceeded();
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -386,12 +412,12 @@ XLogRegisterData(char *data, int len)
  * limited)
  */
 void
-XLogRegisterBufData(uint8 block_id, char *data, int len)
+XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 {
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len <= UINT16_MAX);
 
 	/* find the registered buffer struct */
 	regbuf = &registered_buffers[block_id];
@@ -399,8 +425,16 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
 		elog(ERROR, "no block with id %d registered with WAL insertion",
 			 block_id);
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
+	/*
+	 * Check against max_rdatas; and ensure we don't register more data per
+	 * buffer than can be handled by the physical data format; 
+	 * i.e. that regbuf->rdata_len does not grow beyond what
+	 * XLogRecordBlockHeader->data_length can hold.
+	 */
+	if (num_rdatas >= max_rdatas ||
+		regbuf->rdata_len + len > UINT16_MAX)
+		XLogErrorDataLimitExceeded();
+
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -519,6 +553,12 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, bool *topxid_included)
 {
 	XLogRecData *rdt;
+	/*
+	 * Overflow of total_len is not normally possible, considering that
+	 * this value will be at most 33 RegBufDatas (at UINT16_MAX each)
+	 * plus one MaxXLogRecordSize, which together are still an order of
+	 * magnitude smaller than UINT32_MAX.
+	 */
 	uint32		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
@@ -527,6 +567,9 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	XLogRecord *rechdr;
 	char	   *scratch = hdr_scratch;
 
+	/* ensure that any assembled record can be decoded */
+	Assert(AllocSizeIsValid(DecodeXLogRecordRequiredSpace(MaxXLogRecordSize)));
+
 	/*
 	 * Note: this function can be called multiple times for the same record.
 	 * All the modifications we do to the rdata chains below must handle that.
@@ -756,12 +799,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 
 		if (needs_data)
 		{
+			/*
+			 * When copying to XLogRecordBlockHeader, the length is narrowed
+			 * to an uint16. We double-check that that is still correct.
+			 */
+			Assert(regbuf->rdata_len <= UINT16_MAX);
+
 			/*
 			 * Link the caller-supplied rdata chain for this buffer to the
 			 * overall list.
 			 */
 			bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
-			bkpb.data_length = regbuf->rdata_len;
+			bkpb.data_length = (uint16) regbuf->rdata_len;
 			total_len += regbuf->rdata_len;
 
 			rdt_datas_last->next = regbuf->rdata_head;
@@ -858,6 +907,16 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that the XLogRecord is not too large.
+	 *
+	 * XLogReader machinery is only able to handle records up to a certain
+	 * size (ignoring machine resource limitations), so make sure we will
+	 * not emit records larger than those sizes we advertise we support.
+	 */
+	if (!XLogRecordLengthIsValid(total_len))
+		XLogErrorDataLimitExceeded();
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
diff --git a/src/include/access/xloginsert.h b/src/include/access/xloginsert.h
index c04f77b173..aed4643d1c 100644
--- a/src/include/access/xloginsert.h
+++ b/src/include/access/xloginsert.h
@@ -43,12 +43,12 @@ extern void XLogBeginInsert(void);
 extern void XLogSetRecordFlags(uint8 flags);
 extern XLogRecPtr XLogInsert(RmgrId rmid, uint8 info);
 extern void XLogEnsureRecordSpace(int max_block_id, int ndatas);
-extern void XLogRegisterData(char *data, int len);
+extern void XLogRegisterData(char *data, uint32 len);
 extern void XLogRegisterBuffer(uint8 block_id, Buffer buffer, uint8 flags);
 extern void XLogRegisterBlock(uint8 block_id, RelFileLocator *rlocator,
 							  ForkNumber forknum, BlockNumber blknum, char *page,
 							  uint8 flags);
-extern void XLogRegisterBufData(uint8 block_id, char *data, int len);
+extern void XLogRegisterBufData(uint8 block_id, char *data, uint32 len);
 extern void XLogResetInsertion(void);
 extern bool XLogCheckBufferNeedsBackup(Buffer buffer);
 
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 835151ec92..065bc63a79 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -54,6 +54,19 @@ typedef struct XLogRecord
 
 #define SizeOfXLogRecord	(offsetof(XLogRecord, xl_crc) + sizeof(pg_crc32c))
 
+/*
+ * XLogReader needs to allocate all the data of an xlog record in a single
+ * chunk. This means that a single XLogRecord cannot exceed MaxAllocSize
+ * in length if we ignore any allocation overhead of the XLogReader.
+ *
+ * To accommodate some overhead, hhis MaxXLogRecordSize value allows for
+ * 4MB -1b of allocation overhead in anything that allocates xlog record
+ * data, which should be enough for effectively all purposes.
+ */
+#define MaxXLogRecordSize	(1020 * 1024 * 1024)
+
+#define XLogRecordLengthIsValid(len) ((len) >= 0 && (len) < MaxXLogRecordSize)
+
 /*
  * The high 4 bits in xl_info may be used freely by rmgr. The
  * XLR_SPECIAL_REL_UPDATE and XLR_CHECK_CONSISTENCY bits can be passed by
-- 
2.30.2

From 385ccf1b7d68923009c73db493dbe74be34e305b Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Mon, 19 Dec 2022 12:26:52 +0100
Subject: [PATCH v9] Add protections in xlog record APIs against large numbers
 and overflows.

Before this, it was possible for an extension to create malicious WAL records
that were too large to replay; or that would overflow the xl_tot_len field,
causing potential corruption in WAL record IO ops.

Emitting invalid records was also possible through pg_logical_emit_message(),
which allowed you to emit arbitrary amounts of data up to 2GB, much higher
than the replay limit of approximately 1GB.

This patch adds a limit to the size of an XLogRecord (1020MiB), checks
against overflows, and ensures that the reader infrastructure can read any
max-sized XLogRecords, such that the wal-generating backend will fail when
it attempts to insert unreadable records, instead of that insertion
succeeding but breaking any replication streams.

Author: Matthias van de Meent <boekewurm+postgres@gmail.com>
---
 src/backend/access/transam/xloginsert.c | 59 ++++++++++++++++++++++---
 src/include/access/xlogrecord.h         | 13 ++++++
 2 files changed, 65 insertions(+), 7 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index f811523448..c25431f0e9 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -141,6 +141,17 @@ static XLogRecData *XLogRecordAssemble(RmgrId rmid, uint8 info,
 									   bool *topxid_included);
 static bool XLogCompressBackupBlock(char *page, uint16 hole_offset,
 									uint16 hole_length, char *dest, uint16 *dlen);
+static inline void XLogErrorDataLimitExceeded();
+
+/*
+ * Error due to exceeding the maximum size of a WAL record, or registering
+ * more datas than are being accounted for by the XLog infrastructure.
+ */
+static inline void
+XLogErrorDataLimitExceeded()
+{
+	elog(ERROR, "too much WAL data");
+}
 
 /*
  * Begin constructing a WAL record. This must be called before the
@@ -352,10 +363,25 @@ XLogRegisterData(char *data, uint32 len)
 {
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && XLogRecordLengthIsValid(len));
+
+	/*
+	 * Check against max_rdatas; and ensure we don't fill a record with
+	 * more data than can be replayed. Records are allocated in one chunk
+	 * with some overhead, so ensure XLogRecordLengthIsValid() for that
+	 * size of record.
+	 *
+	 * Additionally, check that we don't accidentally overflow the
+	 * intermediate sum value on 32-bit systems by ensuring that the
+	 * sum of the two inputs is no less than one of the inputs.
+	 */
+	if (num_rdatas >= max_rdatas ||
+#if SIZEOF_SIZE_T == 4
+		 mainrdata_len + len < len ||
+#endif
+		!XLogRecordLengthIsValid((size_t) mainrdata_len + (size_t) len))
+		XLogErrorDataLimitExceeded();
 
-	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -391,7 +417,7 @@ XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
 
-	Assert(begininsert_called);
+	Assert(begininsert_called && len <= UINT16_MAX);
 
 	/* find the registered buffer struct */
 	regbuf = &registered_buffers[block_id];
@@ -400,14 +426,14 @@ XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 			 block_id);
 
 	/*
-	 * Check against max_rdatas and ensure we do not register more data per
+	 * Check against max_rdatas; and ensure we don't register more data per
 	 * buffer than can be handled by the physical data format; i.e. that
 	 * regbuf->rdata_len does not grow beyond what
 	 * XLogRecordBlockHeader->data_length can hold.
 	 */
 	if (num_rdatas >= max_rdatas ||
 		regbuf->rdata_len + len > UINT16_MAX)
-		elog(ERROR, "too much WAL data");
+		XLogErrorDataLimitExceeded();
 
 	rdata = &rdatas[num_rdatas++];
 
@@ -527,6 +553,12 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, bool *topxid_included)
 {
 	XLogRecData *rdt;
+	/*
+	 * Overflow of total_len is not normally possible, considering that
+	 * this value will be at most 33 RegBufDatas (at UINT16_MAX each)
+	 * plus one MaxXLogRecordSize, which together are still an order of
+	 * magnitude smaller than UINT32_MAX.
+	 */
 	uint32		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
@@ -535,6 +567,9 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	XLogRecord *rechdr;
 	char	   *scratch = hdr_scratch;
 
+	/* ensure that any assembled record can be decoded */
+	Assert(AllocSizeIsValid(DecodeXLogRecordRequiredSpace(MaxXLogRecordSize)));
+
 	/*
 	 * Note: this function can be called multiple times for the same record.
 	 * All the modifications we do to the rdata chains below must handle that.
@@ -766,7 +801,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 		{
 			/*
 			 * When copying to XLogRecordBlockHeader, the length is narrowed
-			 * to an uint16.  Double-check that it is still correct.
+			 * to an uint16. We double-check that that is still correct.
 			 */
 			Assert(regbuf->rdata_len <= UINT16_MAX);
 
@@ -872,6 +907,16 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that the XLogRecord is not too large.
+	 *
+	 * XLogReader machinery is only able to handle records up to a certain
+	 * size (ignoring machine resource limitations), so make sure we will
+	 * not emit records larger than those sizes we advertise we support.
+	 */
+	if (!XLogRecordLengthIsValid(total_len))
+		XLogErrorDataLimitExceeded();
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 835151ec92..065bc63a79 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -54,6 +54,19 @@ typedef struct XLogRecord
 
 #define SizeOfXLogRecord	(offsetof(XLogRecord, xl_crc) + sizeof(pg_crc32c))
 
+/*
+ * XLogReader needs to allocate all the data of an xlog record in a single
+ * chunk. This means that a single XLogRecord cannot exceed MaxAllocSize
+ * in length if we ignore any allocation overhead of the XLogReader.
+ *
+ * To accommodate some overhead, hhis MaxXLogRecordSize value allows for
+ * 4MB -1b of allocation overhead in anything that allocates xlog record
+ * data, which should be enough for effectively all purposes.
+ */
+#define MaxXLogRecordSize	(1020 * 1024 * 1024)
+
+#define XLogRecordLengthIsValid(len) ((len) >= 0 && (len) < MaxXLogRecordSize)
+
 /*
  * The high 4 bits in xl_info may be used freely by rmgr. The
  * XLR_SPECIAL_REL_UPDATE and XLR_CHECK_CONSISTENCY bits can be passed by
-- 
2.30.2

From 16b40324a5bc5bbe6e06cbfb86251c907f400151 Mon Sep 17 00:00:00 2001
From: Michael Paquier <michael@paquier.xyz>
Date: Tue, 28 Mar 2023 20:34:31 +0900
Subject: [PATCH v10] Add protections in xlog record APIs against overflows

Before this, it was possible for an extension to create malicious WAL
records that were too large to replay; or that would overflow the
xl_tot_len field, causing potential corruption in WAL record IO ops.

Emitting invalid records was also possible through
pg_logical_emit_message(), which allowed you to emit arbitrary amounts
of data up to 2GB, much higher than the replay limit of approximately
1GB.

This patch adds a limit to the size of an XLogRecord (1020MiB), checks
against overflows, and ensures that the reader infrastructure can read
any max-sized XLogRecords, such that the wal-generating backend will
fail when it attempts to insert unreadable records, instead of that
insertion succeeding but breaking any replication streams.

Author: Matthias van de Meent <boekewurm+postgres@gmail.com>
---
 src/include/access/xlogrecord.h         | 11 ++++
 src/backend/access/transam/xloginsert.c | 73 +++++++++++++++++++++----
 2 files changed, 74 insertions(+), 10 deletions(-)

diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 0d576f7883..0a7b0bb6fc 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -54,6 +54,17 @@ typedef struct XLogRecord
 
 #define SizeOfXLogRecord	(offsetof(XLogRecord, xl_crc) + sizeof(pg_crc32c))
 
+/*
+ * XLogReader needs to allocate all the data of an xlog record in a single
+ * chunk.  This means that a single XLogRecord cannot exceed MaxAllocSize
+ * in length if we ignore any allocation overhead of the XLogReader.
+ *
+ * To accommodate some overhead, this value allows for 4M of allocation
+ * overhead, that should be plenty enough for what
+ * DecodeXLogRecordRequiredSpace() expects as extra.
+ */
+#define XLogRecordMaxSize	(1020 * 1024 * 1024)
+
 /*
  * The high 4 bits in xl_info may be used freely by rmgr. The
  * XLR_SPECIAL_REL_UPDATE and XLR_CHECK_CONSISTENCY bits can be passed by
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 008612e032..86fa6a5276 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -33,6 +33,7 @@
 #include "access/xloginsert.h"
 #include "catalog/pg_control.h"
 #include "common/pg_lzcompress.h"
+#include "common/int.h"
 #include "executor/instrument.h"
 #include "miscadmin.h"
 #include "pg_trace.h"
@@ -351,11 +352,13 @@ void
 XLogRegisterData(char *data, uint32 len)
 {
 	XLogRecData *rdata;
+	uint32		result = 0;
 
 	Assert(begininsert_called);
 
 	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
+		elog(ERROR, "too much WAL data in main chunk: num_rdatas %d, max_rdatas %d",
+			 num_rdatas, max_rdatas);
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -369,7 +372,11 @@ XLogRegisterData(char *data, uint32 len)
 	mainrdata_last->next = rdata;
 	mainrdata_last = rdata;
 
-	mainrdata_len += len;
+	if (pg_add_u32_overflow(mainrdata_len, len, &result))
+		elog(ERROR, "too much WAL data in main chunk: mainrdata_len %u, len %u",
+			 mainrdata_len, len);
+
+	mainrdata_len = result;
 }
 
 /*
@@ -390,6 +397,7 @@ XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 {
 	registered_buffer *regbuf;
 	XLogRecData *rdata;
+	uint32 result = 0;
 
 	Assert(begininsert_called);
 
@@ -405,9 +413,9 @@ XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 	 * regbuf->rdata_len does not grow beyond what
 	 * XLogRecordBlockHeader->data_length can hold.
 	 */
-	if (num_rdatas >= max_rdatas ||
-		regbuf->rdata_len + len > UINT16_MAX)
-		elog(ERROR, "too much WAL data");
+	if (num_rdatas >= max_rdatas)
+		elog(ERROR, "too much WAL data specific to block %u: num_rdatas %u, max_rdatas %u",
+			 block_id, num_rdatas, max_rdatas);
 
 	rdata = &rdatas[num_rdatas++];
 
@@ -416,7 +424,12 @@ XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 
 	regbuf->rdata_tail->next = rdata;
 	regbuf->rdata_tail = rdata;
-	regbuf->rdata_len += len;
+
+	if (pg_add_u32_overflow(regbuf->rdata_len, len, &result) ||
+		regbuf->rdata_len + len > UINT16_MAX)
+		elog(ERROR, "too much WAL data: rdata_len %u, len %u",
+			 regbuf->rdata_len, len);
+	regbuf->rdata_len = result;
 }
 
 /*
@@ -528,6 +541,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 {
 	XLogRecData *rdt;
 	uint32		total_len = 0;
+	uint32		result = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
@@ -759,7 +773,10 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				}
 			}
 
-			total_len += bimg.length;
+			if (pg_add_u32_overflow(total_len, (uint32) bimg.length, &result))
+				elog(ERROR, "too much WAL data: block_id %u, block length %u, total_len %u",
+					 block_id, bimg.length, total_len);
+			total_len = result;
 		}
 
 		if (needs_data)
@@ -776,7 +793,11 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 			 */
 			bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
 			bkpb.data_length = (uint16) regbuf->rdata_len;
-			total_len += regbuf->rdata_len;
+
+			if (pg_add_u32_overflow(total_len, regbuf->rdata_len, &result))
+				elog(ERROR, "too much WAL data: total_len %u, rdata_len %u",
+					 total_len, regbuf->rdata_len);
+			total_len = result;
 
 			rdt_datas_last->next = regbuf->rdata_head;
 			rdt_datas_last = regbuf->rdata_tail;
@@ -852,12 +873,19 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 		}
 		rdt_datas_last->next = mainrdata_head;
 		rdt_datas_last = mainrdata_last;
-		total_len += mainrdata_len;
+
+		if (pg_add_u32_overflow(total_len, mainrdata_len, &result))
+			elog(ERROR, "too much WAL data: total_len %u, mainrdata_len %u",
+				 total_len, mainrdata_len);
+		total_len = result;
 	}
 	rdt_datas_last->next = NULL;
 
 	hdr_rdt.len = (scratch - hdr_scratch);
-	total_len += hdr_rdt.len;
+	if (pg_add_u32_overflow(total_len, hdr_rdt.len, &result))
+		elog(ERROR, "too much WAL data: total_len %u, hdr_rdt.len %u",
+			 total_len, hdr_rdt.len);
+	total_len = result;
 
 	/*
 	 * Calculate CRC of the data
@@ -872,6 +900,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that the XLogRecord is not too large.
+	 *
+	 * XLogReader machinery is only able to handle records up to a certain
+	 * size (ignoring machine resource limitations), so make sure we will
+	 * not emit records larger than those sizes we advertise we support.
+	 * This cap is based on DecodeXLogRecordRequiredSpace().
+	 */
+	if (total_len >= XLogRecordMaxSize)
+		elog(ERROR, "too much WAL data: total_len %u, max %u",
+			 total_len, XLogRecordMaxSize);
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
@@ -1297,6 +1337,19 @@ log_newpage_range(Relation rel, ForkNumber forknum,
 void
 InitXLogInsert(void)
 {
+#ifdef USE_ASSERT_CHECKING
+
+	/*
+	 * Check that any records assembled can be decoded.  This is capped
+	 * based on what XLogReader would require at its maximum bound.
+	 * This code path is called once per backend, more than enough for
+	 * this check.
+	 */
+	size_t max_required = DecodeXLogRecordRequiredSpace(XLogRecordMaxSize);
+
+	Assert(AllocSizeIsValid(max_required));
+#endif
+
 	/* Initialize the working areas */
 	if (xloginsert_cxt == NULL)
 	{
-- 
2.40.0

From 180b6076ecb00cdc2c9b9a7ca1bdd1c7e21f7ee2 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Thu, 30 Mar 2023 20:02:16 +0200
Subject: [PATCH v11 1/2] Create a path for separate xlog record construction
 and insertion

The current record insertion routines are likely to be called inside a
transaction, which means they shouldn't throw errors unless there is
something catastrophically wrong, as it'd cause the server to reboot.
In some cases that is the right thing to do; however, not all WAL records
are created equal, and some cases failing to build the expected WAL
record could easily be resolved by rolling back the transaction.

This prepares the code for XLog record size validation, where the
XLogRecordAssemble() code may throw errors if WAL record size
constraints are violated.
---
 src/backend/access/transam/multixact.c  |   5 +
 src/backend/access/transam/twophase.c   |  58 +++--
 src/backend/access/transam/xact.c       | 298 +++++++++++++-----------
 src/backend/access/transam/xlog.c       |   2 +-
 src/backend/access/transam/xloginsert.c | 122 ++++++++++
 src/backend/commands/sequence.c         |  18 ++
 src/include/access/xact.h               |  64 +++--
 src/include/access/xlog.h               |   1 +
 src/include/access/xloginsert.h         |   3 +
 src/include/access/xlogrecord.h         |  11 +
 10 files changed, 408 insertions(+), 174 deletions(-)

diff --git a/src/backend/access/transam/multixact.c b/src/backend/access/transam/multixact.c
index fe6698d5ff..78d6563df8 100644
--- a/src/backend/access/transam/multixact.c
+++ b/src/backend/access/transam/multixact.c
@@ -832,6 +832,11 @@ MultiXactIdCreateFromMembers(int nmembers, MultiXactMember *members)
 	 * find a more compact representation of this Xlog record -- perhaps all
 	 * the status flags in one XLogRecData, then all the xids in another one?
 	 * Not clear that it's worth the trouble though.
+	 * 
+	 * XXX: large values for nmembers may mean the data would not fit in the
+	 * max xlog record size, in which case this would panic. However, that is
+	 * quite unlikely as such record would log more than 250M members - much
+	 * more than we can currently process efficiently.
 	 */
 	XLogBeginInsert();
 	XLogRegisterData((char *) (&xlrec), SizeOfMultiXactCreate);
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 068e59bec0..f98d79aa29 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -1181,18 +1181,24 @@ EndPrepare(GlobalTransaction gxact)
 	 */
 	XLogEnsureRecordSpace(0, records.num_chunks);
 
-	START_CRIT_SECTION();
-
-	Assert((MyProc->delayChkptFlags & DELAY_CHKPT_START) == 0);
-	MyProc->delayChkptFlags |= DELAY_CHKPT_START;
-
+	/*
+	 * Start preparing the record for insertion. Do so outside the critical
+	 * section to make sure we won't throw errors due to oversized registered
+	 * data.
+	 */
 	XLogBeginInsert();
 	for (record = records.head; record != NULL; record = record->next)
 		XLogRegisterData(record->data, record->len);
 
 	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
+	XLogPrepareInsert(RM_XACT_ID, XLOG_XACT_PREPARE);
+
+	START_CRIT_SECTION();
+
+	Assert((MyProc->delayChkptFlags & DELAY_CHKPT_START) == 0);
+	MyProc->delayChkptFlags |= DELAY_CHKPT_START;
 
-	gxact->prepare_end_lsn = XLogInsert(RM_XACT_ID, XLOG_XACT_PREPARE);
+	gxact->prepare_end_lsn = XLogInsertPrepared();
 
 	if (replorigin)
 	{
@@ -2294,6 +2300,7 @@ RecordTransactionCommitPrepared(TransactionId xid,
 	XLogRecPtr	recptr;
 	TimestampTz committs = GetCurrentTimestamp();
 	bool		replorigin;
+	xl_xact_commit_fields recdata;
 
 	/*
 	 * Are we using the replication origins feature?  Or, in other words, are
@@ -2302,6 +2309,22 @@ RecordTransactionCommitPrepared(TransactionId xid,
 	replorigin = (replorigin_session_origin != InvalidRepOriginId &&
 				  replorigin_session_origin != DoNotReplicateId);
 
+	/*
+	 * We can't assume that the data we're trying to fit in the WAL record is
+	 * going to fit in our XLog record format, due to the arbitrary data we're
+	 * registering. Because xlog record creation will throw an ERROR, we should
+	 * create the record outside the critical section: we don't want to crash
+	 * the backend because of oversized xlog records - we "just" want to error
+	 * out and revert the transaction.
+	 */
+	XactPrepareCommitRecord(committs,
+							nchildren, children, nrels, rels,
+							nstats, stats,
+							ninvalmsgs, invalmsgs,
+							initfileinval,
+							MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
+							xid, gid, &recdata);
+
 	START_CRIT_SECTION();
 
 	/* See notes in RecordTransactionCommit */
@@ -2313,13 +2336,7 @@ RecordTransactionCommitPrepared(TransactionId xid,
 	 * potentially having AccessExclusiveLocks since we don't know whether or
 	 * not they do.
 	 */
-	recptr = XactLogCommitRecord(committs,
-								 nchildren, children, nrels, rels,
-								 nstats, stats,
-								 ninvalmsgs, invalmsgs,
-								 initfileinval,
-								 MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
-								 xid, gid);
+	recptr = XactLogCommitRecord();
 
 
 	if (replorigin)
@@ -2388,6 +2405,7 @@ RecordTransactionAbortPrepared(TransactionId xid,
 {
 	XLogRecPtr	recptr;
 	bool		replorigin;
+	xl_xact_abort_fields recdata;
 
 	/*
 	 * Are we using the replication origins feature?  Or, in other words, are
@@ -2404,6 +2422,13 @@ RecordTransactionAbortPrepared(TransactionId xid,
 		elog(PANIC, "cannot abort transaction %u, it was already committed",
 			 xid);
 
+	XactPrepareAbortRecord(GetCurrentTimestamp(),
+						   nchildren, children,
+						   nrels, rels,
+						   nstats, stats,
+						   MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
+						   xid, gid, &recdata);
+
 	START_CRIT_SECTION();
 
 	/*
@@ -2411,12 +2436,7 @@ RecordTransactionAbortPrepared(TransactionId xid,
 	 * potentially having AccessExclusiveLocks since we don't know whether or
 	 * not they do.
 	 */
-	recptr = XactLogAbortRecord(GetCurrentTimestamp(),
-								nchildren, children,
-								nrels, rels,
-								nstats, stats,
-								MyXactFlags | XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK,
-								xid, gid);
+	recptr = XactLogAbortRecord();
 
 	if (replorigin)
 		/* Move LSNs forward for this replication origin */
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 01b1e0fb8c..79c0448889 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -247,6 +247,10 @@ static TransactionStateData TopTransactionStateData = {
 /*
  * unreportedXids holds XIDs of all subtransactions that have not yet been
  * reported in an XLOG_XACT_ASSIGNMENT record.
+ *
+ * Increasing the size of the unreportedXids cache should be limited to
+ * a reasonable degree, as these xids may need to be fit into a WAL-record,
+ * which have a limited size.
  */
 static int	nUnreportedXids;
 static TransactionId unreportedXids[PGPROC_MAX_CACHED_SUBXIDS];
@@ -1365,6 +1369,7 @@ RecordTransactionCommit(void)
 	else
 	{
 		bool		replorigin;
+		xl_xact_commit_fields recdata;
 
 		/*
 		 * Are we using the replication origins feature?  Or, in other words,
@@ -1376,7 +1381,14 @@ RecordTransactionCommit(void)
 		/*
 		 * Begin commit critical section and insert the commit XLOG record.
 		 */
-
+		XactPrepareCommitRecord(GetCurrentTransactionStopTimestamp(),
+								nchildren, children, nrels, rels,
+								ndroppedstats, droppedstats,
+								nmsgs, invalMessages,
+								RelcacheInitFileInval,
+								MyXactFlags,
+								InvalidTransactionId, NULL /* plain commit */,
+								&recdata);
 		/*
 		 * Mark ourselves as within our "commit critical section".  This
 		 * forces any concurrent checkpoint to wait until we've updated
@@ -1398,13 +1410,7 @@ RecordTransactionCommit(void)
 		START_CRIT_SECTION();
 		MyProc->delayChkptFlags |= DELAY_CHKPT_START;
 
-		XactLogCommitRecord(GetCurrentTransactionStopTimestamp(),
-							nchildren, children, nrels, rels,
-							ndroppedstats, droppedstats,
-							nmsgs, invalMessages,
-							RelcacheInitFileInval,
-							MyXactFlags,
-							InvalidTransactionId, NULL /* plain commit */ );
+		XactLogCommitRecord();
 
 		if (replorigin)
 			/* Move LSNs forward for this replication origin */
@@ -1712,6 +1718,7 @@ RecordTransactionAbort(bool isSubXact)
 	TransactionId *children;
 	TimestampTz xact_time;
 	bool		replorigin;
+	xl_xact_abort_fields recdata;
 
 	/*
 	 * If we haven't been assigned an XID, nobody will care whether we aborted
@@ -1754,10 +1761,6 @@ RecordTransactionAbort(bool isSubXact)
 	nchildren = xactGetCommittedChildren(&children);
 	ndroppedstats = pgstat_get_transactional_drops(false, &droppedstats);
 
-	/* XXX do we really need a critical section here? */
-	START_CRIT_SECTION();
-
-	/* Write the ABORT record */
 	if (isSubXact)
 		xact_time = GetCurrentTimestamp();
 	else
@@ -1765,12 +1768,18 @@ RecordTransactionAbort(bool isSubXact)
 		xact_time = GetCurrentTransactionStopTimestamp();
 	}
 
-	XactLogAbortRecord(xact_time,
-					   nchildren, children,
-					   nrels, rels,
-					   ndroppedstats, droppedstats,
-					   MyXactFlags, InvalidTransactionId,
-					   NULL);
+	XactPrepareAbortRecord(xact_time,
+						   nchildren, children,
+						   nrels, rels,
+						   ndroppedstats, droppedstats,
+						   MyXactFlags, InvalidTransactionId,
+						   NULL, &recdata);
+
+	/* XXX do we really need a critical section here? */
+	START_CRIT_SECTION();
+
+	/* Write the ABORT record */
+	XactLogAbortRecord();
 
 	if (replorigin)
 		/* Move LSNs forward for this replication origin */
@@ -5628,37 +5637,34 @@ xactGetCommittedChildren(TransactionId **ptr)
  *	XLOG support routines
  */
 
-
 /*
- * Log the commit record for a plain or twophase transaction commit.
+ * Prepare the commit record for a plain or twophase transaction commit.
  *
  * A 2pc commit will be emitted when twophase_xid is valid, a plain one
  * otherwise.
+ *
+ * This is separated from writing the record to WAL due to the arbitrary
+ * size of the record potentially resulting in an ERROR whilst constructing
+ * the record, which (when called from critical sections) would panic.
+ * This separation allows us to prepare the insert outside critical sections,
+ * preventing potential crashes.
  */
-XLogRecPtr
-XactLogCommitRecord(TimestampTz commit_time,
-					int nsubxacts, TransactionId *subxacts,
-					int nrels, RelFileLocator *rels,
-					int ndroppedstats, xl_xact_stats_item *droppedstats,
-					int nmsgs, SharedInvalidationMessage *msgs,
-					bool relcacheInval,
-					int xactflags, TransactionId twophase_xid,
-					const char *twophase_gid)
+void
+XactPrepareCommitRecord(TimestampTz commit_time,
+						int nsubxacts, TransactionId *subxacts,
+						int nrels, RelFileLocator *rels,
+						int ndroppedstats, xl_xact_stats_item *droppedstats,
+						int nmsgs, SharedInvalidationMessage *msgs,
+						bool relcacheInval,
+						int xactflags, TransactionId twophase_xid,
+						const char *twophase_gid,
+						xl_xact_commit_fields *recdata)
 {
-	xl_xact_commit xlrec;
-	xl_xact_xinfo xl_xinfo;
-	xl_xact_dbinfo xl_dbinfo;
-	xl_xact_subxacts xl_subxacts;
-	xl_xact_relfilelocators xl_relfilelocators;
-	xl_xact_stats_items xl_dropped_stats;
-	xl_xact_invals xl_invals;
-	xl_xact_twophase xl_twophase;
-	xl_xact_origin xl_origin;
 	uint8		info;
 
-	Assert(CritSectionCount > 0);
+	Assert(CritSectionCount == 0);
 
-	xl_xinfo.xinfo = 0;
+	recdata->xl_xinfo.xinfo = 0;
 
 	/* decide between a plain and 2pc commit */
 	if (!TransactionIdIsValid(twophase_xid))
@@ -5668,21 +5674,21 @@ XactLogCommitRecord(TimestampTz commit_time,
 
 	/* First figure out and collect all the information needed */
 
-	xlrec.xact_time = commit_time;
+	recdata->xlrec.xact_time = commit_time;
 
 	if (relcacheInval)
-		xl_xinfo.xinfo |= XACT_COMPLETION_UPDATE_RELCACHE_FILE;
+		recdata->xl_xinfo.xinfo |= XACT_COMPLETION_UPDATE_RELCACHE_FILE;
 	if (forceSyncCommit)
-		xl_xinfo.xinfo |= XACT_COMPLETION_FORCE_SYNC_COMMIT;
+		recdata->xl_xinfo.xinfo |= XACT_COMPLETION_FORCE_SYNC_COMMIT;
 	if ((xactflags & XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK))
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_AE_LOCKS;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_AE_LOCKS;
 
 	/*
 	 * Check if the caller would like to ask standbys for immediate feedback
 	 * once this commit is applied.
 	 */
 	if (synchronous_commit >= SYNCHRONOUS_COMMIT_REMOTE_APPLY)
-		xl_xinfo.xinfo |= XACT_COMPLETION_APPLY_FEEDBACK;
+		recdata->xl_xinfo.xinfo |= XACT_COMPLETION_APPLY_FEEDBACK;
 
 	/*
 	 * Relcache invalidations requires information about the current database
@@ -5690,145 +5696,154 @@ XactLogCommitRecord(TimestampTz commit_time,
 	 */
 	if (nmsgs > 0 || XLogLogicalInfoActive())
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_DBINFO;
-		xl_dbinfo.dbId = MyDatabaseId;
-		xl_dbinfo.tsId = MyDatabaseTableSpace;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_DBINFO;
+		recdata->xl_dbinfo.dbId = MyDatabaseId;
+		recdata->xl_dbinfo.tsId = MyDatabaseTableSpace;
 	}
 
 	if (nsubxacts > 0)
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_SUBXACTS;
-		xl_subxacts.nsubxacts = nsubxacts;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_SUBXACTS;
+		recdata->xl_subxacts.nsubxacts = nsubxacts;
 	}
 
 	if (nrels > 0)
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_RELFILELOCATORS;
-		xl_relfilelocators.nrels = nrels;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_RELFILELOCATORS;
+		recdata->xl_relfilelocators.nrels = nrels;
 		info |= XLR_SPECIAL_REL_UPDATE;
 	}
 
 	if (ndroppedstats > 0)
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_DROPPED_STATS;
-		xl_dropped_stats.nitems = ndroppedstats;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_DROPPED_STATS;
+		recdata->xl_dropped_stats.nitems = ndroppedstats;
 	}
 
 	if (nmsgs > 0)
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_INVALS;
-		xl_invals.nmsgs = nmsgs;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_INVALS;
+		recdata->xl_invals.nmsgs = nmsgs;
 	}
 
 	if (TransactionIdIsValid(twophase_xid))
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_TWOPHASE;
-		xl_twophase.xid = twophase_xid;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_TWOPHASE;
+		recdata->xl_twophase.xid = twophase_xid;
 		Assert(twophase_gid != NULL);
 
 		if (XLogLogicalInfoActive())
-			xl_xinfo.xinfo |= XACT_XINFO_HAS_GID;
+			recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_GID;
 	}
 
 	/* dump transaction origin information */
 	if (replorigin_session_origin != InvalidRepOriginId)
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_ORIGIN;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_ORIGIN;
 
-		xl_origin.origin_lsn = replorigin_session_origin_lsn;
-		xl_origin.origin_timestamp = replorigin_session_origin_timestamp;
+		recdata->xl_origin.origin_lsn = replorigin_session_origin_lsn;
+		recdata->xl_origin.origin_timestamp = replorigin_session_origin_timestamp;
 	}
 
-	if (xl_xinfo.xinfo != 0)
+	if (recdata->xl_xinfo.xinfo != 0)
 		info |= XLOG_XACT_HAS_INFO;
 
 	/* Then include all the collected data into the commit record. */
 
 	XLogBeginInsert();
 
-	XLogRegisterData((char *) (&xlrec), sizeof(xl_xact_commit));
+	XLogRegisterData((char *) (&recdata->xlrec), sizeof(xl_xact_commit));
 
-	if (xl_xinfo.xinfo != 0)
-		XLogRegisterData((char *) (&xl_xinfo.xinfo), sizeof(xl_xinfo.xinfo));
+	if (recdata->xl_xinfo.xinfo != 0)
+		XLogRegisterData((char *) (&recdata->xl_xinfo.xinfo), sizeof(recdata->xl_xinfo.xinfo));
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_DBINFO)
-		XLogRegisterData((char *) (&xl_dbinfo), sizeof(xl_dbinfo));
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_DBINFO)
+		XLogRegisterData((char *) (&recdata->xl_dbinfo), sizeof(recdata->xl_dbinfo));
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_SUBXACTS)
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_SUBXACTS)
 	{
-		XLogRegisterData((char *) (&xl_subxacts),
+		XLogRegisterData((char *) (&recdata->xl_subxacts),
 						 MinSizeOfXactSubxacts);
 		XLogRegisterData((char *) subxacts,
 						 nsubxacts * sizeof(TransactionId));
 	}
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_RELFILELOCATORS)
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_RELFILELOCATORS)
 	{
-		XLogRegisterData((char *) (&xl_relfilelocators),
+		XLogRegisterData((char *) (&recdata->xl_relfilelocators),
 						 MinSizeOfXactRelfileLocators);
 		XLogRegisterData((char *) rels,
 						 nrels * sizeof(RelFileLocator));
 	}
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_DROPPED_STATS)
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_DROPPED_STATS)
 	{
-		XLogRegisterData((char *) (&xl_dropped_stats),
+		XLogRegisterData((char *) (&recdata->xl_dropped_stats),
 						 MinSizeOfXactStatsItems);
 		XLogRegisterData((char *) droppedstats,
 						 ndroppedstats * sizeof(xl_xact_stats_item));
 	}
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_INVALS)
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_INVALS)
 	{
-		XLogRegisterData((char *) (&xl_invals), MinSizeOfXactInvals);
+		XLogRegisterData((char *) (&recdata->xl_invals), MinSizeOfXactInvals);
 		XLogRegisterData((char *) msgs,
 						 nmsgs * sizeof(SharedInvalidationMessage));
 	}
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_TWOPHASE)
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_TWOPHASE)
 	{
-		XLogRegisterData((char *) (&xl_twophase), sizeof(xl_xact_twophase));
-		if (xl_xinfo.xinfo & XACT_XINFO_HAS_GID)
+		XLogRegisterData((char *) (&recdata->xl_twophase), sizeof(xl_xact_twophase));
+		if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_GID)
 			XLogRegisterData(unconstify(char *, twophase_gid), strlen(twophase_gid) + 1);
 	}
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_ORIGIN)
-		XLogRegisterData((char *) (&xl_origin), sizeof(xl_xact_origin));
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_ORIGIN)
+		XLogRegisterData((char *) (&recdata->xl_origin), sizeof(xl_xact_origin));
 
 	/* we allow filtering by xacts */
 	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
 
-	return XLogInsert(RM_XACT_ID, info);
+	XLogPrepareInsert(RM_XACT_ID, info);
 }
 
 /*
- * Log the commit record for a plain or twophase transaction abort.
+ * Log the prepared XLogCommit record.
+ * See XactPrepareCommitRecord for more info.
+ */
+XLogRecPtr
+XactLogCommitRecord(void)
+{
+	Assert(CritSectionCount > 0);
+	return XLogInsertPrepared();
+}
+
+/*
+ * Prepare the commit record for a plain or twophase transaction abort.
  *
  * A 2pc abort will be emitted when twophase_xid is valid, a plain one
  * otherwise.
+ *
+ * This is separated from writing the record to WAL due to the arbitrary
+ * size of the record potentially resulting in an ERROR whilst constructing
+ * the record, which (when called from critical sections) would panic.
+ * This separation allows us to prepare the insert outside critical sections,
+ * preventing potential crashes.
  */
-XLogRecPtr
-XactLogAbortRecord(TimestampTz abort_time,
-				   int nsubxacts, TransactionId *subxacts,
-				   int nrels, RelFileLocator *rels,
-				   int ndroppedstats, xl_xact_stats_item *droppedstats,
-				   int xactflags, TransactionId twophase_xid,
-				   const char *twophase_gid)
+void
+XactPrepareAbortRecord(TimestampTz abort_time,
+					   int nsubxacts, TransactionId *subxacts,
+					   int nrels, RelFileLocator *rels,
+					   int ndroppedstats, xl_xact_stats_item *droppedstats,
+					   int xactflags, TransactionId twophase_xid,
+					   const char *twophase_gid,
+					   xl_xact_abort_fields *recdata)
 {
-	xl_xact_abort xlrec;
-	xl_xact_xinfo xl_xinfo;
-	xl_xact_subxacts xl_subxacts;
-	xl_xact_relfilelocators xl_relfilelocators;
-	xl_xact_stats_items xl_dropped_stats;
-	xl_xact_twophase xl_twophase;
-	xl_xact_dbinfo xl_dbinfo;
-	xl_xact_origin xl_origin;
-
 	uint8		info;
 
-	Assert(CritSectionCount > 0);
+	Assert(CritSectionCount == 0);
 
-	xl_xinfo.xinfo = 0;
+	recdata->xl_xinfo.xinfo = 0;
 
 	/* decide between a plain and 2pc abort */
 	if (!TransactionIdIsValid(twophase_xid))
@@ -5839,45 +5854,45 @@ XactLogAbortRecord(TimestampTz abort_time,
 
 	/* First figure out and collect all the information needed */
 
-	xlrec.xact_time = abort_time;
+	recdata->xlrec.xact_time = abort_time;
 
 	if ((xactflags & XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK))
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_AE_LOCKS;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_AE_LOCKS;
 
 	if (nsubxacts > 0)
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_SUBXACTS;
-		xl_subxacts.nsubxacts = nsubxacts;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_SUBXACTS;
+		recdata->xl_subxacts.nsubxacts = nsubxacts;
 	}
 
 	if (nrels > 0)
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_RELFILELOCATORS;
-		xl_relfilelocators.nrels = nrels;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_RELFILELOCATORS;
+		recdata->xl_relfilelocators.nrels = nrels;
 		info |= XLR_SPECIAL_REL_UPDATE;
 	}
 
 	if (ndroppedstats > 0)
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_DROPPED_STATS;
-		xl_dropped_stats.nitems = ndroppedstats;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_DROPPED_STATS;
+		recdata->xl_dropped_stats.nitems = ndroppedstats;
 	}
 
 	if (TransactionIdIsValid(twophase_xid))
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_TWOPHASE;
-		xl_twophase.xid = twophase_xid;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_TWOPHASE;
+		recdata->xl_twophase.xid = twophase_xid;
 		Assert(twophase_gid != NULL);
 
 		if (XLogLogicalInfoActive())
-			xl_xinfo.xinfo |= XACT_XINFO_HAS_GID;
+			recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_GID;
 	}
 
 	if (TransactionIdIsValid(twophase_xid) && XLogLogicalInfoActive())
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_DBINFO;
-		xl_dbinfo.dbId = MyDatabaseId;
-		xl_dbinfo.tsId = MyDatabaseTableSpace;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_DBINFO;
+		recdata->xl_dbinfo.dbId = MyDatabaseId;
+		recdata->xl_dbinfo.tsId = MyDatabaseTableSpace;
 	}
 
 	/*
@@ -5886,65 +5901,76 @@ XactLogAbortRecord(TimestampTz abort_time,
 	 */
 	if (replorigin_session_origin != InvalidRepOriginId)
 	{
-		xl_xinfo.xinfo |= XACT_XINFO_HAS_ORIGIN;
+		recdata->xl_xinfo.xinfo |= XACT_XINFO_HAS_ORIGIN;
 
-		xl_origin.origin_lsn = replorigin_session_origin_lsn;
-		xl_origin.origin_timestamp = replorigin_session_origin_timestamp;
+		recdata->xl_origin.origin_lsn = replorigin_session_origin_lsn;
+		recdata->xl_origin.origin_timestamp = replorigin_session_origin_timestamp;
 	}
 
-	if (xl_xinfo.xinfo != 0)
+	if (recdata->xl_xinfo.xinfo != 0)
 		info |= XLOG_XACT_HAS_INFO;
 
 	/* Then include all the collected data into the abort record. */
 
 	XLogBeginInsert();
 
-	XLogRegisterData((char *) (&xlrec), MinSizeOfXactAbort);
+	XLogRegisterData((char *) (&recdata->xlrec), MinSizeOfXactAbort);
 
-	if (xl_xinfo.xinfo != 0)
-		XLogRegisterData((char *) (&xl_xinfo), sizeof(xl_xinfo));
+	if (recdata->xl_xinfo.xinfo != 0)
+		XLogRegisterData((char *) (&recdata->xl_xinfo), sizeof(recdata->xl_xinfo));
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_DBINFO)
-		XLogRegisterData((char *) (&xl_dbinfo), sizeof(xl_dbinfo));
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_DBINFO)
+		XLogRegisterData((char *) (&recdata->xl_dbinfo), sizeof(recdata->xl_dbinfo));
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_SUBXACTS)
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_SUBXACTS)
 	{
-		XLogRegisterData((char *) (&xl_subxacts),
+		XLogRegisterData((char *) (&recdata->xl_subxacts),
 						 MinSizeOfXactSubxacts);
 		XLogRegisterData((char *) subxacts,
 						 nsubxacts * sizeof(TransactionId));
 	}
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_RELFILELOCATORS)
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_RELFILELOCATORS)
 	{
-		XLogRegisterData((char *) (&xl_relfilelocators),
+		XLogRegisterData((char *) (&recdata->xl_relfilelocators),
 						 MinSizeOfXactRelfileLocators);
 		XLogRegisterData((char *) rels,
 						 nrels * sizeof(RelFileLocator));
 	}
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_DROPPED_STATS)
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_DROPPED_STATS)
 	{
-		XLogRegisterData((char *) (&xl_dropped_stats),
+		XLogRegisterData((char *) (&recdata->xl_dropped_stats),
 						 MinSizeOfXactStatsItems);
 		XLogRegisterData((char *) droppedstats,
 						 ndroppedstats * sizeof(xl_xact_stats_item));
 	}
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_TWOPHASE)
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_TWOPHASE)
 	{
-		XLogRegisterData((char *) (&xl_twophase), sizeof(xl_xact_twophase));
-		if (xl_xinfo.xinfo & XACT_XINFO_HAS_GID)
+		XLogRegisterData((char *) (&recdata->xl_twophase), sizeof(xl_xact_twophase));
+		if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_GID)
 			XLogRegisterData(unconstify(char *, twophase_gid), strlen(twophase_gid) + 1);
 	}
 
-	if (xl_xinfo.xinfo & XACT_XINFO_HAS_ORIGIN)
-		XLogRegisterData((char *) (&xl_origin), sizeof(xl_xact_origin));
+	if (recdata->xl_xinfo.xinfo & XACT_XINFO_HAS_ORIGIN)
+		XLogRegisterData((char *) (&recdata->xl_origin), sizeof(xl_xact_origin));
 
 	/* Include the replication origin */
 	XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);
 
-	return XLogInsert(RM_XACT_ID, info);
+	XLogPrepareInsert(RM_XACT_ID, info);
+}
+
+/*
+ * Log the prepared XLogAbort record.
+ * See XactPrepareAbortRecord for more info.
+ */
+XLogRecPtr
+XactLogAbortRecord(void)
+{
+	Assert(CritSectionCount > 0);
+	return XLogInsertPrepared();
 }
 
 /*
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index f9f0f6db8d..b283563f6e 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -821,7 +821,7 @@ XLogInsertRecord(XLogRecData *rdata,
 	doPageWrites = (Insert->fullPageWrites || Insert->runningBackups > 0);
 
 	if (doPageWrites &&
-		(!prevDoPageWrites ||
+		(((!prevDoPageWrites) && ((flags & XLOG_NOPAGEWRITE_USED) != 0)) ||
 		 (fpw_lsn != InvalidXLogRecPtr && fpw_lsn <= RedoRecPtr)))
 	{
 		/*
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 008612e032..134cf32bb4 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -132,6 +132,17 @@ static int	max_rdatas;			/* allocated size */
 
 static bool begininsert_called = false;
 
+/*
+ * Used in split creation / logging of the xlog record, when the caller
+ * wants to protect against >maxsized xlog records errors while inserting the
+ * record in a critical section.
+ */
+static bool prepareinsert_called = false;
+static XLogRecData *prepared_record = NULL;
+static XLogRecPtr preprec_fpw_lsn = InvalidXLogRecPtr;
+static bool	preprec_topxid_included;
+static int	preprec_num_fpw;
+
 /* Memory context to hold the registered buffer and data references. */
 static MemoryContext xloginsert_cxt;
 
@@ -233,6 +244,8 @@ XLogResetInsertion(void)
 	mainrdata_last = (XLogRecData *) &mainrdata_head;
 	curinsert_flags = 0;
 	begininsert_called = false;
+	prepareinsert_called = false;
+	prepared_record = NULL;
 }
 
 /*
@@ -433,6 +446,8 @@ void
 XLogSetRecordFlags(uint8 flags)
 {
 	Assert(begininsert_called);
+	/* don't modify the flags after preparing the record for insertion */
+	Assert(!prepareinsert_called);
 	curinsert_flags |= flags;
 }
 
@@ -506,6 +521,107 @@ XLogInsert(RmgrId rmid, uint8 info)
 	return EndPos;
 }
 
+/*
+ * Prepare to insert an XLOG record having the specified RMID and info bytes,
+ * with the body of the record being the data and buffer references registered
+ * earlier with XLogRegister* calls.
+ * 
+ * The user must later call XLogInsertPrepared() to do the actual insertion,
+ * but once this function returns the user can be assured that the xlog record
+ * itself can be written with the assumed info.
+ */
+void
+XLogPrepareInsert(RmgrId rmid, uint8 info)
+{
+	XLogRecPtr	RedoRecPtr;
+	bool		doPageWrites;
+
+	/* XLogBeginInsert() must have been called. */
+	if (!begininsert_called)
+		elog(ERROR, "XLogBeginInsert was not called");
+	/* ... and you can't prepare the record twice */
+	if (prepareinsert_called)
+		elog(ERROR, "Cannot call XLogPrepareInsert twice");
+
+	prepareinsert_called = true;
+
+	Assert(CritSectionCount == 0);
+	/*
+	 * The caller can set rmgr bits, XLR_SPECIAL_REL_UPDATE and
+	 * XLR_CHECK_CONSISTENCY; the rest are reserved for use by me.
+	 */
+	if ((info & ~(XLR_RMGR_INFO_MASK |
+				  XLR_SPECIAL_REL_UPDATE |
+				  XLR_CHECK_CONSISTENCY)) != 0)
+		elog(PANIC, "invalid xlog info mask %02X", info);
+
+	TRACE_POSTGRESQL_WAL_INSERT(rmid, info);
+
+	/*
+	 * In bootstrap mode, we don't actually log anything but XLOG resources;
+	 * return a phony record pointer.
+	 */
+	if (IsBootstrapProcessingMode() && rmid != RM_XLOG_ID)
+	{
+		prepared_record = NULL;
+		return;
+	}
+
+	preprec_topxid_included = false;
+
+	/*
+	 * Get values needed to decide whether to do full-page writes. Since
+	 * we don't yet have an insertion lock, these could change under us,
+	 * but XLogInsertRecord will recheck them once it has a lock.
+	 */
+	GetFullPageWriteInfo(&RedoRecPtr, &doPageWrites);
+
+	prepared_record = XLogRecordAssemble(rmid, info, RedoRecPtr,
+										 doPageWrites, &preprec_fpw_lsn,
+										 &preprec_num_fpw, &preprec_topxid_included);
+}
+
+/*
+ * XLogInsertPrepared - write the previously prepared record into WAL.
+ * 
+ * Unlike XLogInsert(), this does not retry insertions, but instead returns
+ * InvalidXLogRecPtr when it fails to immediately insert the record (e.g. due
+ * to changes in the FPW settings between prepare and ). 
+ */
+XLogRecPtr
+XLogInsertPrepared(void)
+{
+	XLogRecPtr	EndPos;
+
+	/*
+	 * XLogBeginInsert() must have been called, and a record must have been
+	 * prepared.
+	 */
+	if (!begininsert_called)
+		elog(ERROR, "XLogBeginInsert was not called");
+	if (!prepareinsert_called)
+		elog(ERROR, "XLogPrepareInsert was not called");
+
+	/*
+	 * In bootstrap mode, we don't actually log anything but XLOG resources;
+	 * return a phony record pointer.
+	 */
+	if (IsBootstrapProcessingMode() && prepared_record == NULL)
+	{
+		XLogResetInsertion();
+		EndPos = SizeOfXLogLongPHD; /* start of 1st chkpt record */
+		return EndPos;
+	}
+
+	EndPos = XLogInsertRecord(prepared_record, preprec_fpw_lsn, curinsert_flags,
+							  preprec_num_fpw, preprec_topxid_included);
+
+	Assert(!XLogRecPtrIsInvalid(EndPos));
+	XLogResetInsertion();
+
+	return EndPos;
+}
+
 /*
  * Assemble a WAL record from the registered data and buffers into an
  * XLogRecData chain, ready for insertion with XLogInsertRecord().
@@ -548,6 +664,9 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	rdt_datas_last = &hdr_rdt;
 	hdr_rdt.data = hdr_scratch;
 
+	/* We're rebuilding the record, reset the internal flag */
+	curinsert_flags &= ~XLOG_NOPAGEWRITE_USED;
+
 	/*
 	 * Enforce consistency checks for this record if user is looking for it.
 	 * Do this before at the beginning of this routine to give the possibility
@@ -584,7 +703,10 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 		else if (regbuf->flags & REGBUF_NO_IMAGE)
 			needs_backup = false;
 		else if (!doPageWrites)
+		{
+			curinsert_flags |= XLOG_NOPAGEWRITE_USED;
 			needs_backup = false;
+		}
 		else
 		{
 			/*
diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c
index bfe279cddf..0cc6a25cca 100644
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -431,6 +431,15 @@ fill_seq_fork_with_data(Relation rel, HeapTuple tuple, ForkNumber forkNum)
 		XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
 		XLogRegisterData((char *) tuple->t_data, tuple->t_len);
 
+		/*
+		 * The data in the record can currently not exceed the maximum size of
+		 * a heap tuple by much more than a constant offset, which in turn is
+		 * limited to about BLCKSZ.
+		 * If we ever change this log record's format, be sure to check that
+		 * we can't exceed the maximum size of xlog record while inside this
+		 * critical section, or we'd run the chance of PANICing during this
+		 * insertion.
+		 */
 		recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
 
 		PageSetLSN(page, recptr);
@@ -849,6 +858,15 @@ nextval_internal(Oid relid, bool check_permissions)
 		XLogRegisterData((char *) &xlrec, sizeof(xl_seq_rec));
 		XLogRegisterData((char *) seqdatatuple.t_data, seqdatatuple.t_len);
 
+		/*
+		 * The data in the record can currently not exceed the maximum size of
+		 * a heap tuple by much more than a constant offset, which in turn is
+		 * limited to about BLCKSZ.
+		 * If we ever change this log record's format, be sure to check that
+		 * we can't exceed the maximum size of xlog record while inside this
+		 * critical section, or we'd run the chance of PANICing during this
+		 * insertion.
+		 */
 		recptr = XLogInsert(RM_SEQ_ID, XLOG_SEQ_LOG);
 
 		PageSetLSN(page, recptr);
diff --git a/src/include/access/xact.h b/src/include/access/xact.h
index 7d3b9446e6..7d6a3de0a7 100644
--- a/src/include/access/xact.h
+++ b/src/include/access/xact.h
@@ -425,6 +425,30 @@ typedef struct xl_xact_parsed_abort
 	TimestampTz origin_timestamp;
 } xl_xact_parsed_abort;
 
+/* used to hold the data of the commit record when using prepared xlog insert */
+typedef struct xl_xact_commit_fields {
+	xl_xact_commit xlrec;
+	xl_xact_xinfo xl_xinfo;
+	xl_xact_dbinfo xl_dbinfo;
+	xl_xact_subxacts xl_subxacts;
+	xl_xact_relfilelocators xl_relfilelocators;
+	xl_xact_stats_items xl_dropped_stats;
+	xl_xact_invals xl_invals;
+	xl_xact_twophase xl_twophase;
+	xl_xact_origin xl_origin;
+} xl_xact_commit_fields;
+
+/* used to hold the data of the abort record when using prepared xlog insert */
+typedef struct xl_xact_abort_fields {
+	xl_xact_abort xlrec;
+	xl_xact_xinfo xl_xinfo;
+	xl_xact_subxacts xl_subxacts;
+	xl_xact_relfilelocators xl_relfilelocators;
+	xl_xact_stats_items xl_dropped_stats;
+	xl_xact_twophase xl_twophase;
+	xl_xact_dbinfo xl_dbinfo;
+	xl_xact_origin xl_origin;
+} xl_xact_abort_fields;
 
 /* ----------------
  *		extern definitions
@@ -494,24 +518,28 @@ extern void MarkSubxactTopXidLogged(void);
 
 extern int	xactGetCommittedChildren(TransactionId **ptr);
 
-extern XLogRecPtr XactLogCommitRecord(TimestampTz commit_time,
-									  int nsubxacts, TransactionId *subxacts,
-									  int nrels, RelFileLocator *rels,
-									  int ndroppedstats,
-									  xl_xact_stats_item *droppedstats,
-									  int nmsgs, SharedInvalidationMessage *msgs,
-									  bool relcacheInval,
-									  int xactflags,
-									  TransactionId twophase_xid,
-									  const char *twophase_gid);
-
-extern XLogRecPtr XactLogAbortRecord(TimestampTz abort_time,
-									 int nsubxacts, TransactionId *subxacts,
-									 int nrels, RelFileLocator *rels,
-									 int ndroppedstats,
-									 xl_xact_stats_item *droppedstats,
-									 int xactflags, TransactionId twophase_xid,
-									 const char *twophase_gid);
+extern void XactPrepareCommitRecord(TimestampTz commit_time,
+									int nsubxacts, TransactionId *subxacts,
+									int nrels, RelFileLocator *rels,
+									int ndroppedstats,
+									xl_xact_stats_item *droppedstats,
+									int nmsgs, SharedInvalidationMessage *msgs,
+									bool relcacheInval,
+									int xactflags,
+									TransactionId twophase_xid,
+									const char *twophase_gid,
+									xl_xact_commit_fields *recdata);
+extern XLogRecPtr XactLogCommitRecord(void);
+
+extern void XactPrepareAbortRecord(TimestampTz abort_time,
+								   int nsubxacts, TransactionId *subxacts,
+								   int nrels, RelFileLocator *rels,
+								   int ndroppedstats,
+								   xl_xact_stats_item *droppedstats,
+								   int xactflags, TransactionId twophase_xid,
+								   const char *twophase_gid,
+								   xl_xact_abort_fields *recdata);
+extern XLogRecPtr XactLogAbortRecord(void);
 extern void xact_redo(XLogReaderState *record);
 
 /* xactdesc.c */
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index cfe5409738..c10a3c3679 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -150,6 +150,7 @@ extern PGDLLIMPORT bool XLOG_DEBUG;
  */
 #define XLOG_INCLUDE_ORIGIN		0x01	/* include the replication origin */
 #define XLOG_MARK_UNIMPORTANT	0x02	/* record not important for durability */
+#define XLOG_NOPAGEWRITE_USED	0x04	/* xlog record would've had FPI, if not for settings */
 
 
 /* Checkpoint statistics */
diff --git a/src/include/access/xloginsert.h b/src/include/access/xloginsert.h
index 31785dc578..9d2c8a0893 100644
--- a/src/include/access/xloginsert.h
+++ b/src/include/access/xloginsert.h
@@ -42,6 +42,9 @@
 extern void XLogBeginInsert(void);
 extern void XLogSetRecordFlags(uint8 flags);
 extern XLogRecPtr XLogInsert(RmgrId rmid, uint8 info);
+extern void XLogPrepareInsert(RmgrId rmid, uint8 info);
+extern XLogRecPtr XLogInsertPrepared(void);
+
 extern void XLogEnsureRecordSpace(int max_block_id, int ndatas);
 extern void XLogRegisterData(char *data, uint32 len);
 extern void XLogRegisterBuffer(uint8 block_id, Buffer buffer, uint8 flags);
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 0d576f7883..72bee48c5b 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -54,6 +54,17 @@ typedef struct XLogRecord
 
 #define SizeOfXLogRecord	(offsetof(XLogRecord, xl_crc) + sizeof(pg_crc32c))
 
+/* 
+ * XLogReader needs to allocate all the data of an xlog record in a single
+ * chunk.  This means that a single XLogRecord cannot exceed MaxAllocSize
+ * in length if we ignore any allocation overhead of the XLogReader.
+ *
+ * To accommodate some overhead, this value allows for 4MiB of allocation
+ * overhead, that should be plenty enough for what
+ * DecodeXLogRecordRequiredSpace() expects as extra.
+ */
+#define XLogRecordMaxSize	(1020 * 1024 * 1024)
+
 /*
  * The high 4 bits in xl_info may be used freely by rmgr. The
  * XLR_SPECIAL_REL_UPDATE and XLR_CHECK_CONSISTENCY bits can be passed by
-- 
2.39.0

From ab7f5bd145fe2d21667cb7982db71d3de287b416 Mon Sep 17 00:00:00 2001
From: Matthias van de Meent <boekewurm+postgres@gmail.com>
Date: Wed, 5 Apr 2023 16:11:56 +0200
Subject: [PATCH v11 2/2] Add protections in xlog record APIs against overflows

Before this, it was possible for an extension to create malicious WAL
records that were too large to replay; or that would overflow the
xl_tot_len field, causing potential corruption in WAL record IO ops.

Emitting invalid records was also possible through
pg_logical_emit_message(), which allowed you to emit arbitrary amounts
of data up to 2GB, much higher than the replay limit of approximately
1GB.

This patch adds a limit to the size of an XLogRecord (1020MiB), checks
against overflows, and ensures that the reader infrastructure can read
any max-sized XLogRecords, such that the wal-generating backend will
fail when it attempts to insert unreadable records, instead of that
insertion succeeding but breaking any replication streams.
---
 src/backend/access/transam/xloginsert.c | 49 ++++++++++++++++++++++---
 src/include/access/xlogrecord.h         | 11 ++++++
 2 files changed, 54 insertions(+), 6 deletions(-)

diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 134cf32bb4..6437234994 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -98,7 +98,7 @@ static int	max_registered_block_id = 0;	/* highest block_id + 1 currently
  */
 static XLogRecData *mainrdata_head;
 static XLogRecData *mainrdata_last = (XLogRecData *) &mainrdata_head;
-static uint32 mainrdata_len;	/* total # of bytes in chain */
+static uint64 mainrdata_len;	/* total # of bytes in chain */
 
 /* flags for the in-progress insertion */
 static uint8 curinsert_flags = 0;
@@ -368,7 +368,10 @@ XLogRegisterData(char *data, uint32 len)
 	Assert(begininsert_called);
 
 	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
+		ereport(ERROR,
+				(errmsg_internal("too much WAL data"),
+				 errdetail_internal("%u out of %u data segments are already in use",
+									num_rdatas, max_rdatas)));
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -418,9 +421,16 @@ XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 	 * regbuf->rdata_len does not grow beyond what
 	 * XLogRecordBlockHeader->data_length can hold.
 	 */
-	if (num_rdatas >= max_rdatas ||
-		regbuf->rdata_len + len > UINT16_MAX)
-		elog(ERROR, "too much WAL data");
+	if (num_rdatas >= max_rdatas)
+		ereport(ERROR,
+				(errmsg_internal("too much WAL data"),
+				 errdetail_internal("%u out of %u data segments are already in use",
+									num_rdatas, max_rdatas)));
+	if (regbuf->rdata_len + len > UINT16_MAX || len > UINT16_MAX)
+		ereport(ERROR,
+				(errmsg_internal("too much WAL data"),
+				 errdetail_internal("Registering more than max %u bytes total to block %u: current %uB, adding %uB",
+									UINT16_MAX, block_id, regbuf->rdata_len, len)));
 
 	rdata = &rdatas[num_rdatas++];
 
@@ -643,7 +653,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, bool *topxid_included)
 {
 	XLogRecData *rdt;
-	uint32		total_len = 0;
+	uint64		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
@@ -994,6 +1004,20 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that the XLogRecord is not too large.
+	 *
+	 * XLogReader machinery is only able to handle records up to a certain
+	 * size (ignoring machine resource limitations), so make sure we will
+	 * not emit records larger than those sizes we advertise we support.
+	 * This cap is based on DecodeXLogRecordRequiredSpace().
+	 */
+	if (total_len >= XLogRecordMaxSize)
+		ereport(ERROR,
+				(errmsg_internal("oversized WAL record"),
+				 errdetail_internal("WAL record would be %llu bytes (of max %u bytes); rmid %u flags %u",
+									(unsigned long long) total_len, XLogRecordMaxSize, rmid, info)));
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
@@ -1419,6 +1443,19 @@ log_newpage_range(Relation rel, ForkNumber forknum,
 void
 InitXLogInsert(void)
 {
+#ifdef USE_ASSERT_CHECKING
+
+	/*
+	 * Check that any records assembled can be decoded.  This is capped
+	 * based on what XLogReader would require at its maximum bound.
+	 * This code path is called once per backend, more than enough for
+	 * this check.
+	 */
+	size_t max_required = DecodeXLogRecordRequiredSpace(XLogRecordMaxSize);
+
+	Assert(AllocSizeIsValid(max_required));
+#endif
+
 	/* Initialize the working areas */
 	if (xloginsert_cxt == NULL)
 	{
diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 72bee48c5b..06cca575ea 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -65,6 +65,17 @@ typedef struct XLogRecord
  */
 #define XLogRecordMaxSize	(1020 * 1024 * 1024)
 
+/*
+ * XLogReader needs to allocate all the data of an xlog record in a single
+ * chunk.  This means that a single XLogRecord cannot exceed MaxAllocSize
+ * in length if we ignore any allocation overhead of the XLogReader.
+ *
+ * To accommodate some overhead, this value allows for 4M of allocation
+ * overhead, that should be plenty enough for what
+ * DecodeXLogRecordRequiredSpace() expects as extra.
+ */
+#define XLogRecordMaxSize	(1020 * 1024 * 1024)
+
 /*
  * The high 4 bits in xl_info may be used freely by rmgr. The
  * XLR_SPECIAL_REL_UPDATE and XLR_CHECK_CONSISTENCY bits can be passed by
-- 
2.39.0

From 10eff110176a4988295118b0f17a2a868fb9b96e Mon Sep 17 00:00:00 2001
From: Michael Paquier <michael@paquier.xyz>
Date: Fri, 7 Apr 2023 08:19:24 +0900
Subject: [PATCH v12] Add more protections in WAL record APIs against overflows

This commit adds a limit to the size of an XLogRecord at 1020MiB, based
on a suggestion by Heikki Linnakangas.  This counts for the overhead
needed by the xlogreader when allocating the memory it needs to read a
record in DecodeXLogRecordRequiredSpace().  An assertion based on that
is added to make sure to detect that any additions in the XLogReader
facilities would not cause any overflows.  If that's ever the case, the
upper bound would need to be adjusted.

Before this, it was possible for an external module to create WAL
records large enough to be assembled still not replayable, causing
failures when replaying such WAL records on standbys.  One case
mentioned where this is possible is the in-core function
pg_logical_emit_message(), that allows to emit WAL records with an
arbitrary amount of data up to 2GB, much higher than the replay limit
of approximately 1GB (limit of a palloc).

This commit is a follow-up of ffd1b6b that has added similar protections
for the block-level data.  Here, the checks are extended to the whole
record length.  All the error messages related to overflow checks are
improved to provide more context about what is happening.

Author: Matthias van de Meent
Discussion: https://postgr.es/m/CAEze2WgGiw+LZt+vHf8tWqB_6VxeLsMeoAuod0N=ij1q17n5pw@mail.gmail.com
---
 src/include/access/xlogrecord.h         | 11 +++++
 src/backend/access/transam/xloginsert.c | 60 ++++++++++++++++++++++---
 2 files changed, 64 insertions(+), 7 deletions(-)

diff --git a/src/include/access/xlogrecord.h b/src/include/access/xlogrecord.h
index 0d576f7883..f355e08e1d 100644
--- a/src/include/access/xlogrecord.h
+++ b/src/include/access/xlogrecord.h
@@ -62,6 +62,17 @@ typedef struct XLogRecord
 #define XLR_INFO_MASK			0x0F
 #define XLR_RMGR_INFO_MASK		0xF0
 
+/*
+ * XLogReader needs to allocate all the data of a WAL record in a single
+ * chunk.  This means that a single XLogRecord cannot exceed MaxAllocSize
+ * in length if we ignore any allocation overhead of the XLogReader.
+ *
+ * To accommodate some overhead, this value allows for 4M of allocation
+ * overhead, that should be plenty enough for what
+ * DecodeXLogRecordRequiredSpace() expects as extra.
+ */
+#define XLogRecordMaxSize	(1020 * 1024 * 1024)
+
 /*
  * If a WAL record modifies any relation files, in ways not covered by the
  * usual block references, this flag is set. This is not used for anything
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c
index 008612e032..b71e690c5b 100644
--- a/src/backend/access/transam/xloginsert.c
+++ b/src/backend/access/transam/xloginsert.c
@@ -98,7 +98,7 @@ static int	max_registered_block_id = 0;	/* highest block_id + 1 currently
  */
 static XLogRecData *mainrdata_head;
 static XLogRecData *mainrdata_last = (XLogRecData *) &mainrdata_head;
-static uint32 mainrdata_len;	/* total # of bytes in chain */
+static uint64 mainrdata_len;	/* total # of bytes in chain */
 
 /* flags for the in-progress insertion */
 static uint8 curinsert_flags = 0;
@@ -355,7 +355,10 @@ XLogRegisterData(char *data, uint32 len)
 	Assert(begininsert_called);
 
 	if (num_rdatas >= max_rdatas)
-		elog(ERROR, "too much WAL data");
+		ereport(ERROR,
+				(errmsg_internal("too much WAL data"),
+				 errdetail_internal("%u out of %u data segments are already in use.",
+									num_rdatas, max_rdatas)));
 	rdata = &rdatas[num_rdatas++];
 
 	rdata->data = data;
@@ -405,9 +408,16 @@ XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
 	 * regbuf->rdata_len does not grow beyond what
 	 * XLogRecordBlockHeader->data_length can hold.
 	 */
-	if (num_rdatas >= max_rdatas ||
-		regbuf->rdata_len + len > UINT16_MAX)
-		elog(ERROR, "too much WAL data");
+	if (num_rdatas >= max_rdatas)
+		ereport(ERROR,
+				(errmsg_internal("too much WAL data"),
+				 errdetail_internal("%u out of %u data segments are already in use.",
+									num_rdatas, max_rdatas)));
+	if (regbuf->rdata_len + len > UINT16_MAX || len > UINT16_MAX)
+		ereport(ERROR,
+				(errmsg_internal("too much WAL data"),
+				 errdetail_internal("Registering more than max %u bytes allowed to block %u: current %u bytes, adding %u bytes.",
+									UINT16_MAX, block_id, regbuf->rdata_len, len)));
 
 	rdata = &rdatas[num_rdatas++];
 
@@ -527,7 +537,7 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 				   XLogRecPtr *fpw_lsn, int *num_fpi, bool *topxid_included)
 {
 	XLogRecData *rdt;
-	uint32		total_len = 0;
+	uint64		total_len = 0;
 	int			block_id;
 	pg_crc32c	rdata_crc;
 	registered_buffer *prev_regbuf = NULL;
@@ -841,8 +851,18 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	{
 		if (mainrdata_len > 255)
 		{
+			uint32		mainrdata_len_4b;
+
+			if (mainrdata_len > PG_UINT32_MAX)
+				ereport(ERROR,
+						(errmsg_internal("too much WAL data"),
+						 errdetail_internal("Main data length at %llu bytes for a max of %u bytes.",
+											(unsigned long long) mainrdata_len,
+											PG_UINT32_MAX)));
+
+			mainrdata_len_4b = (uint32) mainrdata_len;
 			*(scratch++) = (char) XLR_BLOCK_ID_DATA_LONG;
-			memcpy(scratch, &mainrdata_len, sizeof(uint32));
+			memcpy(scratch, &mainrdata_len_4b, sizeof(uint32));
 			scratch += sizeof(uint32);
 		}
 		else
@@ -872,6 +892,20 @@ XLogRecordAssemble(RmgrId rmid, uint8 info,
 	for (rdt = hdr_rdt.next; rdt != NULL; rdt = rdt->next)
 		COMP_CRC32C(rdata_crc, rdt->data, rdt->len);
 
+	/*
+	 * Ensure that the XLogRecord is not too large.
+	 *
+	 * XLogReader machinery is only able to handle records up to a certain
+	 * size (ignoring machine resource limitations), so make sure that we will
+	 * not emit records larger than the sizes advertised to be supported.
+	 * This cap is based on DecodeXLogRecordRequiredSpace().
+	 */
+	if (total_len >= XLogRecordMaxSize)
+		ereport(ERROR,
+				(errmsg_internal("oversized WAL record"),
+				 errdetail_internal("WAL record would be %llu bytes (of max %u bytes); rmid %u flags %u.",
+									(unsigned long long) total_len, XLogRecordMaxSize, rmid, info)));
+
 	/*
 	 * Fill in the fields in the record header. Prev-link is filled in later,
 	 * once we know where in the WAL the record will be inserted. The CRC does
@@ -1297,6 +1331,18 @@ log_newpage_range(Relation rel, ForkNumber forknum,
 void
 InitXLogInsert(void)
 {
+#ifdef USE_ASSERT_CHECKING
+
+	/*
+	 * Check that any records assembled can be decoded.  This is capped based
+	 * on what XLogReader would require at its maximum bound.  This code path
+	 * is called once per backend, more than enough for this check.
+	 */
+	size_t		max_required = DecodeXLogRecordRequiredSpace(XLogRecordMaxSize);
+
+	Assert(AllocSizeIsValid(max_required));
+#endif
+
 	/* Initialize the working areas */
 	if (xloginsert_cxt == NULL)
 	{
-- 
2.40.0