From 2246b4834883c236e93ad4132f509a403709f213 Mon Sep 17 00:00:00 2001 From: Eric Seppanen Date: Thu, 15 Apr 2021 14:28:47 -0700 Subject: [PATCH] handle keepalive messages When postgres sends us a keepalive message, send a reply so it doesn't time out and close the connection. The LSN values sent may need to change in the future. Currently we send: write_lsn <= page_cache last_valid_lsn flush_lsn <= page_cache last_valid_lsn apply_lsn <= 0 --- pageserver/src/walreceiver.rs | 31 +++++++++++++++++++++++++++---- 1 file changed, 27 insertions(+), 4 deletions(-) diff --git a/pageserver/src/walreceiver.rs b/pageserver/src/walreceiver.rs index 76471084f7..ecd5cd6ae8 100644 --- a/pageserver/src/walreceiver.rs +++ b/pageserver/src/walreceiver.rs @@ -20,7 +20,7 @@ use crate::PageServerConf; use postgres_protocol::message::backend::ReplicationMessage; use postgres_types::PgLsn; -use tokio_postgres::replication::ReplicationStream; +use tokio_postgres::replication::{PgTimestamp, ReplicationStream}; use tokio_postgres::{Error, NoTls, SimpleQueryMessage, SimpleQueryRow}; @@ -190,9 +190,32 @@ async fn walreceiver_main( } } - ReplicationMessage::PrimaryKeepAlive(_keepalive) => { - trace!("received PrimaryKeepAlive"); - // FIXME: Reply, or the connection will time out + ReplicationMessage::PrimaryKeepAlive(keepalive) => { + let wal_end = keepalive.wal_end(); + let timestamp = keepalive.timestamp(); + let reply_requested: bool = keepalive.reply() != 0; + + trace!( + "received PrimaryKeepAlive(wal_end: {}, timestamp: {} reply: {})", + wal_end, + timestamp, + reply_requested, + ); + if reply_requested { + // TODO: More thought should go into what values are sent here. + let last_lsn = PgLsn::from(pcache.get_last_valid_lsn()); + let write_lsn = last_lsn; + let flush_lsn = last_lsn; + let apply_lsn = PgLsn::INVALID; + let ts = PgTimestamp::now().unwrap(); + const NO_REPLY: u8 = 0u8; + + physical_stream + .as_mut() + .standby_status_update(write_lsn, flush_lsn, apply_lsn, ts, NO_REPLY) + .await + .unwrap(); + } } _ => (), }