fix_pgreceivexlog_problems_v1.patch

*** a/src/bin/pg_basebackup/receivelog.c
--- b/src/bin/pg_basebackup/receivelog.c
***************
*** 38,43 ****
--- 38,47 ----
  #define STREAMING_HEADER_SIZE (1+sizeof(WalDataMessageHeader))
  #define STREAMING_KEEPALIVE_SIZE (1+sizeof(PrimaryKeepaliveMessage))
  
+ /* Kernel FD for an open WAL file */
+ static int		walfile = -1;
+ 
+ 
  /*
   * Open a new WAL file in the specified directory. Store the name
   * (not including the full directory) in namebuf. Assumes there is
***************
*** 96,101 **** open_walfile(XLogRecPtr startpoint, uint32 timeline, char *basedir, char *namebu
--- 100,106 ----
  		{
  			fprintf(stderr, _("%s: could not pad WAL segment %s: %s\n"),
  					progname, fn, strerror(errno));
+ 			free(zerobuf);
  			close(f);
  			unlink(fn);
  			return -1;
***************
*** 120,126 **** open_walfile(XLogRecPtr startpoint, uint32 timeline, char *basedir, char *namebu
   * completed writing the whole segment.
   */
  static bool
! close_walfile(int walfile, char *basedir, char *walname, bool segment_complete)
  {
  	off_t		currpos = lseek(walfile, 0, SEEK_CUR);
  
--- 125,131 ----
   * completed writing the whole segment.
   */
  static bool
! close_walfile(char *basedir, char *walname, bool segment_complete)
  {
  	off_t		currpos = lseek(walfile, 0, SEEK_CUR);
  
***************
*** 142,149 **** close_walfile(int walfile, char *basedir, char *walname, bool segment_complete)
--- 147,156 ----
  	{
  		fprintf(stderr, _("%s: could not close file %s: %s\n"),
  				progname, walname, strerror(errno));
+ 		walfile = -1;
  		return false;
  	}
+ 	walfile = -1;
  
  	/*
  	 * Rename the .partial file only if we've completed writing the whole
***************
*** 270,276 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  	char		current_walfile_name[MAXPGPATH];
  	PGresult   *res;
  	char	   *copybuf = NULL;
- 	int			walfile = -1;
  	int64		last_status = -1;
  	XLogRecPtr	blockpos = InvalidXLogRecPtr;
  
--- 277,282 ----
***************
*** 315,320 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
--- 321,327 ----
  	{
  		fprintf(stderr, _("%s: could not start replication: %s\n"),
  				progname, PQresultErrorMessage(res));
+ 		PQclear(res);
  		return false;
  	}
  	PQclear(res);
***************
*** 341,349 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  		 */
  		if (stream_stop && stream_stop(blockpos, timeline, false))
  		{
! 			if (walfile != -1)
  				/* Potential error message is written by close_walfile */
! 				return close_walfile(walfile, basedir, current_walfile_name, rename_partial);
  			return true;
  		}
  
--- 348,356 ----
  		 */
  		if (stream_stop && stream_stop(blockpos, timeline, false))
  		{
! 			if (walfile != -1 && !close_walfile(basedir, current_walfile_name, rename_partial))
  				/* Potential error message is written by close_walfile */
! 				goto error;
  			return true;
  		}
  
***************
*** 370,376 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  			{
  				fprintf(stderr, _("%s: could not send feedback packet: %s"),
  						progname, PQerrorMessage(conn));
! 				return false;
  			}
  
  			last_status = now;
--- 377,383 ----
  			{
  				fprintf(stderr, _("%s: could not send feedback packet: %s"),
  						progname, PQerrorMessage(conn));
! 				goto error;
  			}
  
  			last_status = now;
***************
*** 421,434 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  			{
  				fprintf(stderr, _("%s: select() failed: %s\n"),
  						progname, strerror(errno));
! 				return false;
  			}
  			/* Else there is actually data on the socket */
  			if (PQconsumeInput(conn) == 0)
  			{
  				fprintf(stderr, _("%s: could not receive data from WAL stream: %s\n"),
  						progname, PQerrorMessage(conn));
! 				return false;
  			}
  			continue;
  		}
--- 428,441 ----
  			{
  				fprintf(stderr, _("%s: select() failed: %s\n"),
  						progname, strerror(errno));
! 				goto error;
  			}
  			/* Else there is actually data on the socket */
  			if (PQconsumeInput(conn) == 0)
  			{
  				fprintf(stderr, _("%s: could not receive data from WAL stream: %s\n"),
  						progname, PQerrorMessage(conn));
! 				goto error;
  			}
  			continue;
  		}
***************
*** 439,445 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  		{
  			fprintf(stderr, _("%s: could not read copy data: %s\n"),
  					progname, PQerrorMessage(conn));
! 			return false;
  		}
  		if (copybuf[0] == 'k')
  		{
--- 446,452 ----
  		{
  			fprintf(stderr, _("%s: could not read copy data: %s\n"),
  					progname, PQerrorMessage(conn));
! 			goto error;
  		}
  		if (copybuf[0] == 'k')
  		{
***************
*** 451,457 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  			{
  				fprintf(stderr, _("%s: keepalive message is incorrect size: %d\n"),
  						progname, r);
! 				return false;
  			}
  			continue;
  		}
--- 458,464 ----
  			{
  				fprintf(stderr, _("%s: keepalive message is incorrect size: %d\n"),
  						progname, r);
! 				goto error;
  			}
  			continue;
  		}
***************
*** 459,471 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  		{
  			fprintf(stderr, _("%s: unrecognized streaming header: \"%c\"\n"),
  					progname, copybuf[0]);
! 			return false;
  		}
  		if (r < STREAMING_HEADER_SIZE + 1)
  		{
  			fprintf(stderr, _("%s: streaming header too small: %d\n"),
  					progname, r);
! 			return false;
  		}
  
  		/* Extract WAL location for this block */
--- 466,478 ----
  		{
  			fprintf(stderr, _("%s: unrecognized streaming header: \"%c\"\n"),
  					progname, copybuf[0]);
! 			goto error;
  		}
  		if (r < STREAMING_HEADER_SIZE + 1)
  		{
  			fprintf(stderr, _("%s: streaming header too small: %d\n"),
  					progname, r);
! 			goto error;
  		}
  
  		/* Extract WAL location for this block */
***************
*** 483,489 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  			{
  				fprintf(stderr, _("%s: received xlog record for offset %u with no file open\n"),
  						progname, xlogoff);
! 				return false;
  			}
  		}
  		else
--- 490,496 ----
  			{
  				fprintf(stderr, _("%s: received xlog record for offset %u with no file open\n"),
  						progname, xlogoff);
! 				goto error;
  			}
  		}
  		else
***************
*** 494,500 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  			{
  				fprintf(stderr, _("%s: got WAL data offset %08x, expected %08x\n"),
  						progname, xlogoff, (int) lseek(walfile, 0, SEEK_CUR));
! 				return false;
  			}
  		}
  
--- 501,507 ----
  			{
  				fprintf(stderr, _("%s: got WAL data offset %08x, expected %08x\n"),
  						progname, xlogoff, (int) lseek(walfile, 0, SEEK_CUR));
! 				goto error;
  			}
  		}
  
***************
*** 520,526 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  									   basedir, current_walfile_name);
  				if (walfile == -1)
  					/* Error logged by open_walfile */
! 					return false;
  			}
  
  			if (write(walfile,
--- 527,533 ----
  									   basedir, current_walfile_name);
  				if (walfile == -1)
  					/* Error logged by open_walfile */
! 					goto error;
  			}
  
  			if (write(walfile,
***************
*** 532,538 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  						bytes_to_write,
  						current_walfile_name,
  						strerror(errno));
! 				return false;
  			}
  
  			/* Write was successful, advance our position */
--- 539,545 ----
  						bytes_to_write,
  						current_walfile_name,
  						strerror(errno));
! 				goto error;
  			}
  
  			/* Write was successful, advance our position */
***************
*** 544,554 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  			/* Did we reach the end of a WAL segment? */
  			if (blockpos % XLOG_SEG_SIZE == 0)
  			{
! 				if (!close_walfile(walfile, basedir, current_walfile_name, false))
  					/* Error message written in close_walfile() */
! 					return false;
  
- 				walfile = -1;
  				xlogoff = 0;
  
  				if (stream_stop != NULL)
--- 551,560 ----
  			/* Did we reach the end of a WAL segment? */
  			if (blockpos % XLOG_SEG_SIZE == 0)
  			{
! 				if (walfile != -1 && !close_walfile(basedir, current_walfile_name, false))
  					/* Error message written in close_walfile() */
! 					goto error;
  
  				xlogoff = 0;
  
  				if (stream_stop != NULL)
***************
*** 577,584 **** ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
  	{
  		fprintf(stderr, _("%s: unexpected termination of replication stream: %s\n"),
  				progname, PQresultErrorMessage(res));
! 		return false;
  	}
  	PQclear(res);
  	return true;
  }
--- 583,604 ----
  	{
  		fprintf(stderr, _("%s: unexpected termination of replication stream: %s\n"),
  				progname, PQresultErrorMessage(res));
! 		goto error;
  	}
  	PQclear(res);
+ 
+ 	if (copybuf != NULL)
+ 		PQfreemem(copybuf);
+ 	if (walfile != -1 && close(walfile) != 0)
+ 		fprintf(stderr, _("%s: could not close file %s: %s\n"),
+ 				progname, current_walfile_name, strerror(errno));
  	return true;
+ 
+ error:
+ 	if (copybuf != NULL)
+ 		PQfreemem(copybuf);
+ 	if (walfile != -1 && close(walfile) != 0)
+ 		fprintf(stderr, _("%s: could not close file %s: %s\n"),
+ 				progname, current_walfile_name, strerror(errno));
+ 	return false;
  }
*** a/src/bin/pg_basebackup/streamutil.c
--- b/src/bin/pg_basebackup/streamutil.c
***************
*** 143,148 **** GetConnection(void)
--- 143,160 ----
  
  		tmpconn = PQconnectdbParams(keywords, values, true);
  
+ 		/*
+ 		 * If there is too little memory even to allocate the PGconn object
+ 		 * and PQconnectdbParams returns NULL, we call exit(1) directly.
+ 		 */
+ 		if (!tmpconn)
+ 		{
+ 			fprintf(stderr, _("%s: could not connect to server\n"),
+ 					progname);
+ 			PQfinish(tmpconn);
+ 			exit(1);
+ 		}
+ 
  		if (PQstatus(tmpconn) == CONNECTION_BAD &&
  			PQconnectionNeedsPassword(tmpconn) &&
  			dbgetpassword != -1)
***************
*** 154,161 **** GetConnection(void)
  
  		if (PQstatus(tmpconn) != CONNECTION_OK)
  		{
! 			fprintf(stderr, _("%s: could not connect to server: %s"),
  					progname, PQerrorMessage(tmpconn));
  			return NULL;
  		}
  
--- 166,176 ----
  
  		if (PQstatus(tmpconn) != CONNECTION_OK)
  		{
! 			fprintf(stderr, _("%s: could not connect to server: %s\n"),
  					progname, PQerrorMessage(tmpconn));
+ 			PQfinish(tmpconn);
+ 			free(values);
+ 			free(keywords);
  			return NULL;
  		}