walrcv_avoid_data_loss_v1.patch

*** a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
--- b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
***************
*** 49,55 **** static char *recvBuf = NULL;
  static bool libpqrcv_connect(char *conninfo, XLogRecPtr startpoint);
  static bool libpqrcv_receive(int timeout, unsigned char *type,
  				 char **buffer, int *len);
! static void libpqrcv_send(const char *buffer, int nbytes);
  static void libpqrcv_disconnect(void);
  
  /* Prototypes for private functions */
--- 49,55 ----
  static bool libpqrcv_connect(char *conninfo, XLogRecPtr startpoint);
  static bool libpqrcv_receive(int timeout, unsigned char *type,
  				 char **buffer, int *len);
! static bool libpqrcv_send(const char *buffer, int nbytes);
  static void libpqrcv_disconnect(void);
  
  /* Prototypes for private functions */
***************
*** 404,417 **** libpqrcv_receive(int timeout, unsigned char *type, char **buffer, int *len)
  /*
   * Send a message to XLOG stream.
   *
!  * ereports on error.
   */
! static void
  libpqrcv_send(const char *buffer, int nbytes)
  {
  	if (PQputCopyData(streamConn, buffer, nbytes) <= 0 ||
  		PQflush(streamConn))
! 		ereport(ERROR,
! 				(errmsg("could not send data to WAL stream: %s",
! 						PQerrorMessage(streamConn))));
  }
--- 404,428 ----
  /*
   * Send a message to XLOG stream.
   *
!  * Return true if successfully, false otherwise.
   */
! static bool
  libpqrcv_send(const char *buffer, int nbytes)
  {
+ 	/*
+ 	 * Even if we could not send data to WAL stream, we don't emit ERROR
+ 	 * just yet. There might be still data available in the receive buffer. We
+ 	 * emit ERROR after all available data have been received and flushed to
+ 	 * disk. Otherwise, such outstanding data would be lost.
+ 	 *
+ 	 * XXX: Should the result of PQerrorMessage() be returned so that the
+ 	 * caller can report it? This doesn't seem worthwhile because in most cases,
+ 	 * before reporting that, we will get another error and emit ERROR while
+ 	 * trying to process all available data.
+ 	 */
  	if (PQputCopyData(streamConn, buffer, nbytes) <= 0 ||
  		PQflush(streamConn))
! 		return false;
! 
! 	return true;
  }
*** a/src/backend/replication/walreceiver.c
--- b/src/backend/replication/walreceiver.c
***************
*** 96,101 **** static struct
--- 96,104 ----
  static StandbyReplyMessage reply_message;
  static StandbyHSFeedbackMessage feedback_message;
  
+ /* Did previous attempt to send data to WAL stream fail? */
+ static bool walrcv_send_error = false;
+ 
  /*
   * About SIGTERM handling:
   *
***************
*** 328,333 **** WalReceiverMain(void)
--- 331,345 ----
  		else
  		{
  			/*
+ 			 * If we didn't receive anything new after we had failed to send data
+ 			 * to WAL stream, we can guarantee that there is no data available in
+ 			 * the receive buffer. So we at last emit ERROR.
+ 			 */
+ 			if (walrcv_send_error)
+ 				ereport(ERROR,
+ 						(errmsg("could not send data to WAL stream")));
+ 
+ 			/*
  			 * We didn't receive anything new, but send a status update to the
  			 * master anyway, to report any progress in applying WAL.
  			 */
***************
*** 627,632 **** XLogWalRcvSendReply(void)
--- 639,651 ----
  									   wal_receiver_status_interval * 1000))
  		return;
  
+ 	/*
+ 	 * If previous attempt to send data to WAL stream has failed, there is
+ 	 * nothing to do here because this attempt would fail again.
+ 	 */
+ 	if (walrcv_send_error)
+ 		return;
+ 
  	/* Construct a new message */
  	reply_message.write = LogstreamResult.Write;
  	reply_message.flush = LogstreamResult.Flush;
***************
*** 641,647 **** XLogWalRcvSendReply(void)
  	/* Prepend with the message type and send it. */
  	buf[0] = 'r';
  	memcpy(&buf[1], &reply_message, sizeof(StandbyReplyMessage));
! 	walrcv_send(buf, sizeof(StandbyReplyMessage) + 1);
  }
  
  /*
--- 660,666 ----
  	/* Prepend with the message type and send it. */
  	buf[0] = 'r';
  	memcpy(&buf[1], &reply_message, sizeof(StandbyReplyMessage));
! 	walrcv_send_error = !walrcv_send(buf, sizeof(StandbyReplyMessage) + 1);
  }
  
  /*
***************
*** 682,687 **** XLogWalRcvSendHSFeedback(void)
--- 701,714 ----
  		return;
  
  	/*
+ 	 * If previous attempt to send data to WAL stream has failed, there is
+ 	 * nothing to do here because this attempt would fail again. Check this
+ 	 * after the interval has expired to reduce number of calls.
+ 	 */
+ 	if (walrcv_send_error)
+ 		return;
+ 
+ 	/*
  	 * Make the expensive call to get the oldest xmin once we are certain
  	 * everything else has been checked.
  	 */
***************
*** 709,713 **** XLogWalRcvSendHSFeedback(void)
  	/* Prepend with the message type and send it. */
  	buf[0] = 'h';
  	memcpy(&buf[1], &feedback_message, sizeof(StandbyHSFeedbackMessage));
! 	walrcv_send(buf, sizeof(StandbyHSFeedbackMessage) + 1);
  }
--- 736,740 ----
  	/* Prepend with the message type and send it. */
  	buf[0] = 'h';
  	memcpy(&buf[1], &feedback_message, sizeof(StandbyHSFeedbackMessage));
! 	walrcv_send_error = !walrcv_send(buf, sizeof(StandbyHSFeedbackMessage) + 1);
  }
*** a/src/include/replication/walreceiver.h
--- b/src/include/replication/walreceiver.h
***************
*** 96,102 **** typedef bool (*walrcv_receive_type) (int timeout, unsigned char *type,
  												 char **buffer, int *len);
  extern PGDLLIMPORT walrcv_receive_type walrcv_receive;
  
! typedef void (*walrcv_send_type) (const char *buffer, int nbytes);
  extern PGDLLIMPORT walrcv_send_type walrcv_send;
  
  typedef void (*walrcv_disconnect_type) (void);
--- 96,102 ----
  												 char **buffer, int *len);
  extern PGDLLIMPORT walrcv_receive_type walrcv_receive;
  
! typedef bool (*walrcv_send_type) (const char *buffer, int nbytes);
  extern PGDLLIMPORT walrcv_send_type walrcv_send;
  
  typedef void (*walrcv_disconnect_type) (void);