Possible gaps/garbage in the output of XLOG reader

diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
new file mode 100644
index 3a86f34..764a579
*** a/src/backend/access/transam/xlogreader.c
--- b/src/backend/access/transam/xlogreader.c
*************** XLogReadRecord(XLogReaderState *state, X
*** 345,351 ****
  		goto err;
  	}
  
! 	len = XLOG_BLCKSZ - RecPtr % XLOG_BLCKSZ;
  	if (total_len > len)
  	{
  		/* Need to reassemble record */
--- 345,365 ----
  		goto err;
  	}
  
! 	/* Bytes of the current record residing on the current page. */
! 	len = Min(XLOG_BLCKSZ - targetRecOff, total_len);
! 
! 	/*
! 	 * Nothing beyond the record header is guaranteed to be in state->readBuf
! 	 * so far.
! 	 */
! 	if (readOff < targetRecOff + len)
! 	{
! 		readOff = ReadPageInternal(state, targetPagePtr, targetRecOff + len);
! 
! 		if (readOff < 0)
! 			goto err;
! 	}
! 
  	if (total_len > len)
  	{
  		/* Need to reassemble record */
*************** XLogReadRecord(XLogReaderState *state, X
*** 354,362 ****
  		char	   *buffer;
  		uint32		gotlen;
  
  		/* Copy the first fragment of the record from the first page. */
! 		memcpy(state->readRecordBuf,
! 			   state->readBuf + RecPtr % XLOG_BLCKSZ, len);
  		buffer = state->readRecordBuf + len;
  		gotlen = len;
  
--- 368,378 ----
  		char	   *buffer;
  		uint32		gotlen;
  
+ 		Assert(readOff == targetRecOff + len);
+ 		Assert(readOff == XLOG_BLCKSZ);
+ 
  		/* Copy the first fragment of the record from the first page. */
! 		memcpy(state->readRecordBuf, state->readBuf + targetRecOff, len);
  		buffer = state->readRecordBuf + len;
  		gotlen = len;
  
*************** XLogReadRecord(XLogReaderState *state, X
*** 445,464 ****
  	}
  	else
  	{
! 		/* Wait for the record data to become available */
! 		readOff = ReadPageInternal(state, targetPagePtr,
! 								   Min(targetRecOff + total_len, XLOG_BLCKSZ));
! 		if (readOff < 0)
! 			goto err;
  
  		/* Record does not cross a page boundary */
  		if (!ValidXLogRecord(state, record, RecPtr))
  			goto err;
  
! 		state->EndRecPtr = RecPtr + MAXALIGN(total_len);
  
  		state->ReadRecPtr = RecPtr;
! 		memcpy(state->readRecordBuf, record, total_len);
  	}
  
  	/*
--- 461,480 ----
  	}
  	else
  	{
! 		/*
! 		 * The whole record is on the current page and should have been
! 		 * fetched into readBuf.
! 		 */
! 		Assert(readOff >= targetRecOff + len);
  
  		/* Record does not cross a page boundary */
  		if (!ValidXLogRecord(state, record, RecPtr))
  			goto err;
  
! 		state->EndRecPtr = RecPtr + MAXALIGN(len);
  
  		state->ReadRecPtr = RecPtr;
! 		memcpy(state->readRecordBuf, record, len);
  	}
  
  	/*

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
new file mode 100644
index e42b828..e7b56b6
*** a/src/backend/access/transam/xlog.c
--- b/src/backend/access/transam/xlog.c
*************** retry:
*** 11648,11654 ****
  	}
  
  	pgstat_report_wait_start(WAIT_EVENT_WAL_READ);
! 	if (read(readFile, readBuf, XLOG_BLCKSZ) != XLOG_BLCKSZ)
  	{
  		char		fname[MAXFNAMELEN];
  
--- 11648,11654 ----
  	}
  
  	pgstat_report_wait_start(WAIT_EVENT_WAL_READ);
! 	if (read(readFile, readBuf, readLen) != readLen)
  	{
  		char		fname[MAXFNAMELEN];
  
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c
new file mode 100644
index 3a86f34..12d38f0
*** a/src/backend/access/transam/xlogreader.c
--- b/src/backend/access/transam/xlogreader.c
*************** XLogReadRecord(XLogReaderState *state, X
*** 354,359 ****
--- 354,365 ----
  		char	   *buffer;
  		uint32		gotlen;
  
+ 		/*
+ 		 * Since we're going to copy the rest of the current page, the whole
+ 		 * page should be in readBuf.
+ 		 */
+ 		Assert(readOff == XLOG_BLCKSZ);
+ 
  		/* Copy the first fragment of the record from the first page. */
  		memcpy(state->readRecordBuf,
  			   state->readBuf + RecPtr % XLOG_BLCKSZ, len);
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
new file mode 100644
index 130ecd5..9f2d40b
*** a/src/backend/replication/walsender.c
--- b/src/backend/replication/walsender.c
***************
*** 103,109 ****
   * because signals are checked only between messages.  128kB (with
   * default 8k blocks) seems like a reasonable guess for now.
   */
! #define MAX_SEND_SIZE (XLOG_BLCKSZ * 16)
  
  /* Array of WalSnds in shared memory */
  WalSndCtlData *WalSndCtl = NULL;
--- 103,109 ----
   * because signals are checked only between messages.  128kB (with
   * default 8k blocks) seems like a reasonable guess for now.
   */
! #define MAX_SEND_SIZE (XLOG_BLCKSZ / 2)
  
  /* Array of WalSnds in shared memory */
  WalSndCtlData *WalSndCtl = NULL;
*************** XLogSendPhysical(void)
*** 2688,2694 ****
  	else
  	{
  		/* round down to page boundary. */
! 		endptr -= (endptr % XLOG_BLCKSZ);
  		WalSndCaughtUp = false;
  	}
  
--- 2688,2694 ----
  	else
  	{
  		/* round down to page boundary. */
! 		//endptr -= (endptr % XLOG_BLCKSZ);
  		WalSndCaughtUp = false;
  	}