Teach pg_basebackup and pg_receivexlog to reply to server keepalives.

Without this, the connection will be killed after timeout if
wal_sender_timeout is set in the server.

Original patch by Amit Kapila, modified by me to fit recent changes in the
code.

diff --git a/doc/src/sgml/ref/pg_basebackup.sgml b/doc/src/sgml/ref/pg_basebackup.sgml
index a951d6b0f043961b998b1c76851b9f6f3e6e22b0..0bc3ca27b168da03794e653fa9127dfde0220807 100644 (file)
--- a/doc/src/sgml/ref/pg_basebackup.sgml
+++ b/doc/src/sgml/ref/pg_basebackup.sgml
@@ -377,10 +377,10 @@ PostgreSQL documentation
       <listitem>
        <para>
         Specifies the number of seconds between status packets sent back to the
-        server. This is required when streaming the transaction log (using
-        <literal>--xlog=stream</literal>) if replication timeout is configured
-        on the server, and allows for easier monitoring. A value of zero disables
-        the status updates completely. The default value is 10 seconds.
+        server. This allows for easier monitoring of the progress from server.
+        A value of zero disables the periodic status updates completely,
+        although an update will still be sent when requested by the server, to
+        avoid timeout disconnect. The default value is 10 seconds.
        </para>
       </listitem>
      </varlistentry>

diff --git a/doc/src/sgml/ref/pg_receivexlog.sgml b/doc/src/sgml/ref/pg_receivexlog.sgml
index 7f62fd9e6156133e238d9c01a7efbcba78738114..d06dd1f171c6ec3c8f2b524166b3f08ee8b0b2e2 100644 (file)
--- a/doc/src/sgml/ref/pg_receivexlog.sgml
+++ b/doc/src/sgml/ref/pg_receivexlog.sgml
@@ -155,9 +155,10 @@ PostgreSQL documentation
       <listitem>
        <para>
         Specifies the number of seconds between status packets sent back to the
-        server. This is required if replication timeout is configured on the
-        server, and allows for easier monitoring. A value of zero disables the
-        status updates completely. The default value is 10 seconds.
+        server. This allows for easier monitoring of the progress from server.
+        A value of zero disables the periodic status updates completely,
+        although an update will still be sent when requested by the server, to
+        avoid timeout disconnect. The default value is 10 seconds.
        </para>
       </listitem>
      </varlistentry>

diff --git a/src/bin/pg_basebackup/receivelog.c b/src/bin/pg_basebackup/receivelog.c
index f653650ffb14c2dd28a42146a571b41b728a5224..de82ff54d8e762215624c80bf59565feb60004fe 100644 (file)
--- a/src/bin/pg_basebackup/receivelog.c
+++ b/src/bin/pg_basebackup/receivelog.c
@@ -287,7 +287,7 @@ recvint64(char *buf)
  * Send a Standby Status Update message to server.
  */
 static bool
-sendFeedback(PGconn *conn, XLogRecPtr blockpos, int64 now)
+sendFeedback(PGconn *conn, XLogRecPtr blockpos, int64 now, bool replyRequested)
 {
        char            replybuf[1 + 8 + 8 + 8 + 8 + 1];
        int             len = 0;
@@ -302,7 +302,7 @@ sendFeedback(PGconn *conn, XLogRecPtr blockpos, int64 now)
        len += 8;
        sendint64(now, &replybuf[len]);                                 /* sendTime */
        len += 8;
-       replybuf[len] = 0;                                                              /* replyRequested */
+       replybuf[len] = replyRequested ? 1 : 0;                 /* replyRequested */
        len += 1;
 
        if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
@@ -413,6 +413,7 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
                int                     bytes_left;
                int                     bytes_written;
                int64           now;
+               int                     hdr_len;
 
                if (copybuf != NULL)
                {
@@ -441,7 +442,7 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
                                                                                        standby_message_timeout))
                {
                        /* Time to send feedback! */
-                       if (!sendFeedback(conn, blockpos, now))
+                       if (!sendFeedback(conn, blockpos, now, false))
                                goto error;
                        last_status = now;
                }
@@ -520,10 +521,34 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
                /* Check the message type. */
                if (copybuf[0] == 'k')
                {
+                       int             pos;
+                       bool    replyRequested;
+
                        /*
-                        * keepalive message, sent in 9.2 and newer. We just ignore this
-                        * message completely, but need to skip past it in the stream.
+                        * Parse the keepalive message, enclosed in the CopyData message.
+                        * We just check if the server requested a reply, and ignore the
+                        * rest.
                         */
+                       pos = 1;        /* skip msgtype 'k' */
+                       pos += 8;       /* skip walEnd */
+                       pos += 8;       /* skip sendTime */
+
+                       if (r < pos + 1)
+                       {
+                               fprintf(stderr, _("%s: streaming header too small: %d\n"),
+                                               progname, r);
+                               goto error;
+                       }
+                       replyRequested = copybuf[pos];
+
+                       /* If the server requested an immediate reply, send one. */
+                       if (replyRequested)
+                       {
+                               now = localGetCurrentTimestamp();
+                               if (!sendFeedback(conn, blockpos, now, false))
+                                       goto error;
+                               last_status = now;
+                       }
                        continue;
                }
                else if (copybuf[0] != 'w')
@@ -538,8 +563,11 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
                 * message. We only need the WAL location field (dataStart), the rest
                 * of the header is ignored.
                 */
-#define STREAMING_HEADER_SIZE (1 /* msgtype */ + 8 /* dataStart */ + 8 /* walEnd */ + 8 /* sendTime */)
-               if (r < STREAMING_HEADER_SIZE + 1)
+               hdr_len = 1;    /* msgtype 'w' */
+               hdr_len += 8;   /* dataStart */
+               hdr_len += 8;   /* walEnd */
+               hdr_len += 8;   /* sendTime */
+               if (r < hdr_len + 1)
                {
                        fprintf(stderr, _("%s: streaming header too small: %d\n"),
                                        progname, r);
@@ -578,7 +606,7 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
                        }
                }
 
-               bytes_left = r - STREAMING_HEADER_SIZE;
+               bytes_left = r - hdr_len;
                bytes_written = 0;
 
                while (bytes_left)
@@ -604,7 +632,7 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
                        }
 
                        if (write(walfile,
-                                         copybuf + STREAMING_HEADER_SIZE + bytes_written,
+                                         copybuf + hdr_len + bytes_written,
                                          bytes_to_write) != bytes_to_write)
                        {
                                fprintf(stderr,