Fix pg_basebackup for keepalive messages

Teach pg_basebackup in streaming mode to deal with keepalive messages.
Also change the order of checks to complain at the message rather than
block size when a new message is introduced.

In passing, switch to using sizeof() instead of hardcoded sizes for
WAL protocol structs.

diff --git a/src/bin/pg_basebackup/receivelog.c b/src/bin/pg_basebackup/receivelog.c
index c18db4fd042cc7096acabc357419b95d035f3f57..e3a0e92d363930017034c8d7744bb6ab02df3fc0 100644 (file)
--- a/src/bin/pg_basebackup/receivelog.c
+++ b/src/bin/pg_basebackup/receivelog.c
@@ -33,8 +33,9 @@
 #include <unistd.h>
 
 
-/* Size of the streaming replication protocol header */
-#define STREAMING_HEADER_SIZE (1+8+8+8)
+/* Size of the streaming replication protocol headers */
+#define STREAMING_HEADER_SIZE (1+sizeof(WalDataMessageHeader))
+#define STREAMING_KEEPALIVE_SIZE (1+sizeof(PrimaryKeepaliveMessage))
 
 const XLogRecPtr InvalidXLogRecPtr = {0, 0};
 
@@ -374,18 +375,33 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysi
                                        progname, PQerrorMessage(conn));
                        return false;
                }
-               if (r < STREAMING_HEADER_SIZE + 1)
+               if (copybuf[0] == 'k')
                {
-                       fprintf(stderr, _("%s: streaming header too small: %i\n"),
-                                       progname, r);
-                       return false;
+                       /*
+                        * keepalive message, sent in 9.2 and newer. We just ignore
+                        * this message completely, but need to forward past it
+                        * in our reading.
+                        */
+                       if (r != STREAMING_KEEPALIVE_SIZE)
+                       {
+                               fprintf(stderr, _("%s: keepalive message is incorrect size: %i\n"),
+                                               progname, r);
+                               return false;
+                       }
+                       continue;
                }
-               if (copybuf[0] != 'w')
+               else if (copybuf[0] != 'w')
                {
                        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: %i\n"),
+                                       progname, r);
+                       return false;
+               }
 
                /* Extract WAL location for this block */
                memcpy(&blockpos, copybuf + 1, 8);