LCOV - code coverage report
Current view: top level - safekeeper/src - send_wal.rs (source / functions) Coverage Total Hit
Test: 5fe7fa8d483b39476409aee736d6d5e32728bfac.info Lines: 17.2 % 696 120
Test Date: 2025-03-12 16:10:49 Functions: 17.6 % 91 16

            Line data    Source code
       1              : //! This module implements the streaming side of replication protocol, starting
       2              : //! with the "START_REPLICATION" message, and registry of walsenders.
       3              : 
       4              : use std::cmp::{max, min};
       5              : use std::net::SocketAddr;
       6              : use std::sync::Arc;
       7              : use std::time::Duration;
       8              : 
       9              : use anyhow::{Context as AnyhowContext, bail};
      10              : use bytes::Bytes;
      11              : use futures::FutureExt;
      12              : use itertools::Itertools;
      13              : use parking_lot::Mutex;
      14              : use postgres_backend::{CopyStreamHandlerEnd, PostgresBackend, PostgresBackendReader, QueryError};
      15              : use postgres_ffi::{MAX_SEND_SIZE, TimestampTz, get_current_timestamp};
      16              : use pq_proto::{BeMessage, WalSndKeepAlive, XLogDataBody};
      17              : use safekeeper_api::Term;
      18              : use safekeeper_api::models::{
      19              :     HotStandbyFeedback, INVALID_FULL_TRANSACTION_ID, ReplicationFeedback, StandbyFeedback,
      20              :     StandbyReply,
      21              : };
      22              : use tokio::io::{AsyncRead, AsyncWrite};
      23              : use tokio::sync::watch::Receiver;
      24              : use tokio::time::timeout;
      25              : use tracing::*;
      26              : use utils::bin_ser::BeSer;
      27              : use utils::failpoint_support;
      28              : use utils::lsn::Lsn;
      29              : use utils::pageserver_feedback::PageserverFeedback;
      30              : use utils::postgres_client::PostgresClientProtocol;
      31              : 
      32              : use crate::handler::SafekeeperPostgresHandler;
      33              : use crate::metrics::{RECEIVED_PS_FEEDBACKS, WAL_READERS};
      34              : use crate::receive_wal::WalReceivers;
      35              : use crate::safekeeper::TermLsn;
      36              : use crate::send_interpreted_wal::{
      37              :     Batch, InterpretedWalReader, InterpretedWalReaderHandle, InterpretedWalSender,
      38              : };
      39              : use crate::timeline::WalResidentTimeline;
      40              : use crate::wal_reader_stream::StreamingWalReader;
      41              : use crate::wal_storage::WalReader;
      42              : 
      43              : // See: https://www.postgresql.org/docs/13/protocol-replication.html
      44              : const HOT_STANDBY_FEEDBACK_TAG_BYTE: u8 = b'h';
      45              : const STANDBY_STATUS_UPDATE_TAG_BYTE: u8 = b'r';
      46              : // neon extension of replication protocol
      47              : const NEON_STATUS_UPDATE_TAG_BYTE: u8 = b'z';
      48              : 
      49              : /// WalSenders registry. Timeline holds it (wrapped in Arc).
      50              : pub struct WalSenders {
      51              :     mutex: Mutex<WalSendersShared>,
      52              :     walreceivers: Arc<WalReceivers>,
      53              : }
      54              : 
      55              : pub struct WalSendersTimelineMetricValues {
      56              :     pub ps_feedback_counter: u64,
      57              :     pub last_ps_feedback: PageserverFeedback,
      58              :     pub interpreted_wal_reader_tasks: usize,
      59              : }
      60              : 
      61              : impl WalSenders {
      62            5 :     pub fn new(walreceivers: Arc<WalReceivers>) -> Arc<WalSenders> {
      63            5 :         Arc::new(WalSenders {
      64            5 :             mutex: Mutex::new(WalSendersShared::new()),
      65            5 :             walreceivers,
      66            5 :         })
      67            5 :     }
      68              : 
      69              :     /// Register new walsender. Returned guard provides access to the slot and
      70              :     /// automatically deregisters in Drop.
      71            0 :     fn register(self: &Arc<WalSenders>, walsender_state: WalSenderState) -> WalSenderGuard {
      72            0 :         let slots = &mut self.mutex.lock().slots;
      73              :         // find empty slot or create new one
      74            0 :         let pos = if let Some(pos) = slots.iter().position(|s| s.is_none()) {
      75            0 :             slots[pos] = Some(walsender_state);
      76            0 :             pos
      77              :         } else {
      78            0 :             let pos = slots.len();
      79            0 :             slots.push(Some(walsender_state));
      80            0 :             pos
      81              :         };
      82            0 :         WalSenderGuard {
      83            0 :             id: pos,
      84            0 :             walsenders: self.clone(),
      85            0 :         }
      86            0 :     }
      87              : 
      88            0 :     fn create_or_update_interpreted_reader<
      89            0 :         FUp: FnOnce(&Arc<InterpretedWalReaderHandle>) -> anyhow::Result<()>,
      90            0 :         FNew: FnOnce() -> InterpretedWalReaderHandle,
      91            0 :     >(
      92            0 :         self: &Arc<WalSenders>,
      93            0 :         id: WalSenderId,
      94            0 :         start_pos: Lsn,
      95            0 :         max_delta_for_fanout: Option<u64>,
      96            0 :         update: FUp,
      97            0 :         create: FNew,
      98            0 :     ) -> anyhow::Result<()> {
      99            0 :         let state = &mut self.mutex.lock();
     100            0 : 
     101            0 :         let mut selected_interpreted_reader = None;
     102            0 :         for slot in state.slots.iter().flatten() {
     103            0 :             if let WalSenderState::Interpreted(slot_state) = slot {
     104            0 :                 if let Some(ref interpreted_reader) = slot_state.interpreted_wal_reader {
     105            0 :                     let select = match (interpreted_reader.current_position(), max_delta_for_fanout)
     106              :                     {
     107            0 :                         (Some(pos), Some(max_delta)) => {
     108            0 :                             let delta = pos.0.abs_diff(start_pos.0);
     109            0 :                             delta <= max_delta
     110              :                         }
     111              :                         // Reader is not active
     112            0 :                         (None, _) => false,
     113              :                         // Gating fanout by max delta is disabled.
     114              :                         // Attach to any active reader.
     115            0 :                         (_, None) => true,
     116              :                     };
     117              : 
     118            0 :                     if select {
     119            0 :                         selected_interpreted_reader = Some(interpreted_reader.clone());
     120            0 :                         break;
     121            0 :                     }
     122            0 :                 }
     123            0 :             }
     124              :         }
     125              : 
     126            0 :         let slot = state.get_slot_mut(id);
     127            0 :         let slot_state = match slot {
     128            0 :             WalSenderState::Interpreted(s) => s,
     129            0 :             WalSenderState::Vanilla(_) => unreachable!(),
     130              :         };
     131              : 
     132            0 :         let selected_or_new = match selected_interpreted_reader {
     133            0 :             Some(selected) => {
     134            0 :                 update(&selected)?;
     135            0 :                 selected
     136              :             }
     137            0 :             None => Arc::new(create()),
     138              :         };
     139              : 
     140            0 :         slot_state.interpreted_wal_reader = Some(selected_or_new);
     141            0 : 
     142            0 :         Ok(())
     143            0 :     }
     144              : 
     145              :     /// Get state of all walsenders.
     146            0 :     pub fn get_all_public(self: &Arc<WalSenders>) -> Vec<safekeeper_api::models::WalSenderState> {
     147            0 :         self.mutex
     148            0 :             .lock()
     149            0 :             .slots
     150            0 :             .iter()
     151            0 :             .flatten()
     152            0 :             .map(|state| match state {
     153            0 :                 WalSenderState::Vanilla(s) => {
     154            0 :                     safekeeper_api::models::WalSenderState::Vanilla(s.clone())
     155              :                 }
     156            0 :                 WalSenderState::Interpreted(s) => {
     157            0 :                     safekeeper_api::models::WalSenderState::Interpreted(s.public_state.clone())
     158              :                 }
     159            0 :             })
     160            0 :             .collect()
     161            0 :     }
     162              : 
     163              :     /// Get LSN of the most lagging pageserver receiver. Return None if there are no
     164              :     /// active walsenders.
     165            0 :     pub fn laggard_lsn(self: &Arc<WalSenders>) -> Option<Lsn> {
     166            0 :         self.mutex
     167            0 :             .lock()
     168            0 :             .slots
     169            0 :             .iter()
     170            0 :             .flatten()
     171            0 :             .filter_map(|s| match s.get_feedback() {
     172            0 :                 ReplicationFeedback::Pageserver(feedback) => Some(feedback.last_received_lsn),
     173            0 :                 ReplicationFeedback::Standby(_) => None,
     174            0 :             })
     175            0 :             .min()
     176            0 :     }
     177              : 
     178              :     /// Returns total counter of pageserver feedbacks received and last feedback.
     179            0 :     pub fn info_for_metrics(self: &Arc<WalSenders>) -> WalSendersTimelineMetricValues {
     180            0 :         let shared = self.mutex.lock();
     181            0 : 
     182            0 :         let interpreted_wal_reader_tasks = shared
     183            0 :             .slots
     184            0 :             .iter()
     185            0 :             .filter_map(|ss| match ss {
     186            0 :                 Some(WalSenderState::Interpreted(int)) => int.interpreted_wal_reader.as_ref(),
     187            0 :                 Some(WalSenderState::Vanilla(_)) => None,
     188            0 :                 None => None,
     189            0 :             })
     190            0 :             .unique_by(|reader| Arc::as_ptr(reader))
     191            0 :             .count();
     192            0 : 
     193            0 :         WalSendersTimelineMetricValues {
     194            0 :             ps_feedback_counter: shared.ps_feedback_counter,
     195            0 :             last_ps_feedback: shared.last_ps_feedback,
     196            0 :             interpreted_wal_reader_tasks,
     197            0 :         }
     198            0 :     }
     199              : 
     200              :     /// Get aggregated hot standby feedback (we send it to compute).
     201          620 :     pub fn get_hotstandby(self: &Arc<WalSenders>) -> StandbyFeedback {
     202          620 :         self.mutex.lock().agg_standby_feedback
     203          620 :     }
     204              : 
     205              :     /// Record new pageserver feedback, update aggregated values.
     206            0 :     fn record_ps_feedback(self: &Arc<WalSenders>, id: WalSenderId, feedback: &PageserverFeedback) {
     207            0 :         let mut shared = self.mutex.lock();
     208            0 :         *shared.get_slot_mut(id).get_mut_feedback() = ReplicationFeedback::Pageserver(*feedback);
     209            0 :         shared.last_ps_feedback = *feedback;
     210            0 :         shared.ps_feedback_counter += 1;
     211            0 :         drop(shared);
     212            0 : 
     213            0 :         RECEIVED_PS_FEEDBACKS.inc();
     214            0 : 
     215            0 :         // send feedback to connected walproposers
     216            0 :         self.walreceivers.broadcast_pageserver_feedback(*feedback);
     217            0 :     }
     218              : 
     219              :     /// Record standby reply.
     220            0 :     fn record_standby_reply(self: &Arc<WalSenders>, id: WalSenderId, reply: &StandbyReply) {
     221            0 :         let mut shared = self.mutex.lock();
     222            0 :         let slot = shared.get_slot_mut(id);
     223            0 :         debug!(
     224            0 :             "Record standby reply: ts={} apply_lsn={}",
     225              :             reply.reply_ts, reply.apply_lsn
     226              :         );
     227            0 :         match &mut slot.get_mut_feedback() {
     228            0 :             ReplicationFeedback::Standby(sf) => sf.reply = *reply,
     229              :             ReplicationFeedback::Pageserver(_) => {
     230            0 :                 *slot.get_mut_feedback() = ReplicationFeedback::Standby(StandbyFeedback {
     231            0 :                     reply: *reply,
     232            0 :                     hs_feedback: HotStandbyFeedback::empty(),
     233            0 :                 })
     234              :             }
     235              :         }
     236            0 :     }
     237              : 
     238              :     /// Record hot standby feedback, update aggregated value.
     239            0 :     fn record_hs_feedback(self: &Arc<WalSenders>, id: WalSenderId, feedback: &HotStandbyFeedback) {
     240            0 :         let mut shared = self.mutex.lock();
     241            0 :         let slot = shared.get_slot_mut(id);
     242            0 :         match &mut slot.get_mut_feedback() {
     243            0 :             ReplicationFeedback::Standby(sf) => sf.hs_feedback = *feedback,
     244              :             ReplicationFeedback::Pageserver(_) => {
     245            0 :                 *slot.get_mut_feedback() = ReplicationFeedback::Standby(StandbyFeedback {
     246            0 :                     reply: StandbyReply::empty(),
     247            0 :                     hs_feedback: *feedback,
     248            0 :                 })
     249              :             }
     250              :         }
     251            0 :         shared.update_reply_feedback();
     252            0 :     }
     253              : 
     254              :     /// Get remote_consistent_lsn reported by the pageserver. Returns None if
     255              :     /// client is not pageserver.
     256            0 :     pub fn get_ws_remote_consistent_lsn(self: &Arc<WalSenders>, id: WalSenderId) -> Option<Lsn> {
     257            0 :         let shared = self.mutex.lock();
     258            0 :         let slot = shared.get_slot(id);
     259            0 :         match slot.get_feedback() {
     260            0 :             ReplicationFeedback::Pageserver(feedback) => Some(feedback.remote_consistent_lsn),
     261            0 :             _ => None,
     262              :         }
     263            0 :     }
     264              : 
     265              :     /// Unregister walsender.
     266            0 :     fn unregister(self: &Arc<WalSenders>, id: WalSenderId) {
     267            0 :         let mut shared = self.mutex.lock();
     268            0 :         shared.slots[id] = None;
     269            0 :         shared.update_reply_feedback();
     270            0 :     }
     271              : }
     272              : 
     273              : struct WalSendersShared {
     274              :     // aggregated over all walsenders value
     275              :     agg_standby_feedback: StandbyFeedback,
     276              :     // last feedback ever received from any pageserver, empty if none
     277              :     last_ps_feedback: PageserverFeedback,
     278              :     // total counter of pageserver feedbacks received
     279              :     ps_feedback_counter: u64,
     280              :     slots: Vec<Option<WalSenderState>>,
     281              : }
     282              : 
     283              : /// Safekeeper internal definitions of wal sender state
     284              : ///
     285              : /// As opposed to [`safekeeper_api::models::WalSenderState`] these struct may
     286              : /// include state that we don not wish to expose to the public api.
     287              : #[derive(Debug, Clone)]
     288              : pub(crate) enum WalSenderState {
     289              :     Vanilla(VanillaWalSenderInternalState),
     290              :     Interpreted(InterpretedWalSenderInternalState),
     291              : }
     292              : 
     293              : type VanillaWalSenderInternalState = safekeeper_api::models::VanillaWalSenderState;
     294              : 
     295              : #[derive(Debug, Clone)]
     296              : pub(crate) struct InterpretedWalSenderInternalState {
     297              :     public_state: safekeeper_api::models::InterpretedWalSenderState,
     298              :     interpreted_wal_reader: Option<Arc<InterpretedWalReaderHandle>>,
     299              : }
     300              : 
     301              : impl WalSenderState {
     302            0 :     fn get_addr(&self) -> &SocketAddr {
     303            0 :         match self {
     304            0 :             WalSenderState::Vanilla(state) => &state.addr,
     305            0 :             WalSenderState::Interpreted(state) => &state.public_state.addr,
     306              :         }
     307            0 :     }
     308              : 
     309            4 :     fn get_feedback(&self) -> &ReplicationFeedback {
     310            4 :         match self {
     311            4 :             WalSenderState::Vanilla(state) => &state.feedback,
     312            0 :             WalSenderState::Interpreted(state) => &state.public_state.feedback,
     313              :         }
     314            4 :     }
     315              : 
     316            0 :     fn get_mut_feedback(&mut self) -> &mut ReplicationFeedback {
     317            0 :         match self {
     318            0 :             WalSenderState::Vanilla(state) => &mut state.feedback,
     319            0 :             WalSenderState::Interpreted(state) => &mut state.public_state.feedback,
     320              :         }
     321            0 :     }
     322              : }
     323              : 
     324              : impl WalSendersShared {
     325            7 :     fn new() -> Self {
     326            7 :         WalSendersShared {
     327            7 :             agg_standby_feedback: StandbyFeedback::empty(),
     328            7 :             last_ps_feedback: PageserverFeedback::empty(),
     329            7 :             ps_feedback_counter: 0,
     330            7 :             slots: Vec::new(),
     331            7 :         }
     332            7 :     }
     333              : 
     334              :     /// Get content of provided id slot, it must exist.
     335            0 :     fn get_slot(&self, id: WalSenderId) -> &WalSenderState {
     336            0 :         self.slots[id].as_ref().expect("walsender doesn't exist")
     337            0 :     }
     338              : 
     339              :     /// Get mut content of provided id slot, it must exist.
     340            0 :     fn get_slot_mut(&mut self, id: WalSenderId) -> &mut WalSenderState {
     341            0 :         self.slots[id].as_mut().expect("walsender doesn't exist")
     342            0 :     }
     343              : 
     344              :     /// Update aggregated hot standy and normal reply feedbacks. We just take min of valid xmins
     345              :     /// and ts.
     346            2 :     fn update_reply_feedback(&mut self) {
     347            2 :         let mut agg = HotStandbyFeedback::empty();
     348            2 :         let mut reply_agg = StandbyReply::empty();
     349            4 :         for ws_state in self.slots.iter().flatten() {
     350            4 :             if let ReplicationFeedback::Standby(standby_feedback) = ws_state.get_feedback() {
     351            4 :                 let hs_feedback = standby_feedback.hs_feedback;
     352            4 :                 // doing Option math like op1.iter().chain(op2.iter()).min()
     353            4 :                 // would be nicer, but we serialize/deserialize this struct
     354            4 :                 // directly, so leave as is for now
     355            4 :                 if hs_feedback.xmin != INVALID_FULL_TRANSACTION_ID {
     356            2 :                     if agg.xmin != INVALID_FULL_TRANSACTION_ID {
     357            1 :                         agg.xmin = min(agg.xmin, hs_feedback.xmin);
     358            1 :                     } else {
     359            1 :                         agg.xmin = hs_feedback.xmin;
     360            1 :                     }
     361            2 :                     agg.ts = max(agg.ts, hs_feedback.ts);
     362            2 :                 }
     363            4 :                 if hs_feedback.catalog_xmin != INVALID_FULL_TRANSACTION_ID {
     364            0 :                     if agg.catalog_xmin != INVALID_FULL_TRANSACTION_ID {
     365            0 :                         agg.catalog_xmin = min(agg.catalog_xmin, hs_feedback.catalog_xmin);
     366            0 :                     } else {
     367            0 :                         agg.catalog_xmin = hs_feedback.catalog_xmin;
     368            0 :                     }
     369            0 :                     agg.ts = max(agg.ts, hs_feedback.ts);
     370            4 :                 }
     371            4 :                 let reply = standby_feedback.reply;
     372            4 :                 if reply.write_lsn != Lsn::INVALID {
     373            0 :                     if reply_agg.write_lsn != Lsn::INVALID {
     374            0 :                         reply_agg.write_lsn = Lsn::min(reply_agg.write_lsn, reply.write_lsn);
     375            0 :                     } else {
     376            0 :                         reply_agg.write_lsn = reply.write_lsn;
     377            0 :                     }
     378            4 :                 }
     379            4 :                 if reply.flush_lsn != Lsn::INVALID {
     380            0 :                     if reply_agg.flush_lsn != Lsn::INVALID {
     381            0 :                         reply_agg.flush_lsn = Lsn::min(reply_agg.flush_lsn, reply.flush_lsn);
     382            0 :                     } else {
     383            0 :                         reply_agg.flush_lsn = reply.flush_lsn;
     384            0 :                     }
     385            4 :                 }
     386            4 :                 if reply.apply_lsn != Lsn::INVALID {
     387            0 :                     if reply_agg.apply_lsn != Lsn::INVALID {
     388            0 :                         reply_agg.apply_lsn = Lsn::min(reply_agg.apply_lsn, reply.apply_lsn);
     389            0 :                     } else {
     390            0 :                         reply_agg.apply_lsn = reply.apply_lsn;
     391            0 :                     }
     392            4 :                 }
     393            4 :                 if reply.reply_ts != 0 {
     394            0 :                     if reply_agg.reply_ts != 0 {
     395            0 :                         reply_agg.reply_ts = TimestampTz::min(reply_agg.reply_ts, reply.reply_ts);
     396            0 :                     } else {
     397            0 :                         reply_agg.reply_ts = reply.reply_ts;
     398            0 :                     }
     399            4 :                 }
     400            0 :             }
     401              :         }
     402            2 :         self.agg_standby_feedback = StandbyFeedback {
     403            2 :             reply: reply_agg,
     404            2 :             hs_feedback: agg,
     405            2 :         };
     406            2 :     }
     407              : }
     408              : 
     409              : // id of the occupied slot in WalSenders to access it (and save in the
     410              : // WalSenderGuard). We could give Arc directly to the slot, but there is not
     411              : // much sense in that as values aggregation which is performed on each feedback
     412              : // receival iterates over all walsenders.
     413              : pub type WalSenderId = usize;
     414              : 
     415              : /// Scope guard to access slot in WalSenders registry and unregister from it in
     416              : /// Drop.
     417              : pub struct WalSenderGuard {
     418              :     id: WalSenderId,
     419              :     walsenders: Arc<WalSenders>,
     420              : }
     421              : 
     422              : impl WalSenderGuard {
     423            0 :     pub fn id(&self) -> WalSenderId {
     424            0 :         self.id
     425            0 :     }
     426              : 
     427            0 :     pub fn walsenders(&self) -> &Arc<WalSenders> {
     428            0 :         &self.walsenders
     429            0 :     }
     430              : }
     431              : 
     432              : impl Drop for WalSenderGuard {
     433            0 :     fn drop(&mut self) {
     434            0 :         self.walsenders.unregister(self.id);
     435            0 :     }
     436              : }
     437              : 
     438              : impl SafekeeperPostgresHandler {
     439              :     /// Wrapper around handle_start_replication_guts handling result. Error is
     440              :     /// handled here while we're still in walsender ttid span; with API
     441              :     /// extension, this can probably be moved into postgres_backend.
     442            0 :     pub async fn handle_start_replication<IO: AsyncRead + AsyncWrite + Unpin + Send>(
     443            0 :         &mut self,
     444            0 :         pgb: &mut PostgresBackend<IO>,
     445            0 :         start_pos: Lsn,
     446            0 :         term: Option<Term>,
     447            0 :     ) -> Result<(), QueryError> {
     448            0 :         let tli = self
     449            0 :             .global_timelines
     450            0 :             .get(self.ttid)
     451            0 :             .map_err(|e| QueryError::Other(e.into()))?;
     452            0 :         let residence_guard = tli.wal_residence_guard().await?;
     453              : 
     454            0 :         if let Err(end) = self
     455            0 :             .handle_start_replication_guts(pgb, start_pos, term, residence_guard)
     456            0 :             .await
     457              :         {
     458            0 :             let info = tli.get_safekeeper_info(&self.conf).await;
     459              :             // Log the result and probably send it to the client, closing the stream.
     460            0 :             pgb.handle_copy_stream_end(end)
     461            0 :             .instrument(info_span!("", term=%info.term, last_log_term=%info.last_log_term, flush_lsn=%Lsn(info.flush_lsn), commit_lsn=%Lsn(info.flush_lsn)))
     462            0 :             .await;
     463            0 :         }
     464            0 :         Ok(())
     465            0 :     }
     466              : 
     467            0 :     pub async fn handle_start_replication_guts<IO: AsyncRead + AsyncWrite + Unpin + Send>(
     468            0 :         &mut self,
     469            0 :         pgb: &mut PostgresBackend<IO>,
     470            0 :         start_pos: Lsn,
     471            0 :         term: Option<Term>,
     472            0 :         tli: WalResidentTimeline,
     473            0 :     ) -> Result<(), CopyStreamHandlerEnd> {
     474            0 :         let appname = self.appname.clone();
     475              : 
     476              :         // Use a guard object to remove our entry from the timeline when we are done.
     477            0 :         let ws_guard = match self.protocol() {
     478            0 :             PostgresClientProtocol::Vanilla => Arc::new(tli.get_walsenders().register(
     479            0 :                 WalSenderState::Vanilla(VanillaWalSenderInternalState {
     480            0 :                     ttid: self.ttid,
     481            0 :                     addr: *pgb.get_peer_addr(),
     482            0 :                     conn_id: self.conn_id,
     483            0 :                     appname: self.appname.clone(),
     484            0 :                     feedback: ReplicationFeedback::Pageserver(PageserverFeedback::empty()),
     485            0 :                 }),
     486            0 :             )),
     487            0 :             PostgresClientProtocol::Interpreted { .. } => Arc::new(tli.get_walsenders().register(
     488            0 :                 WalSenderState::Interpreted(InterpretedWalSenderInternalState {
     489            0 :                     public_state: safekeeper_api::models::InterpretedWalSenderState {
     490            0 :                         ttid: self.ttid,
     491            0 :                         shard: self.shard.unwrap(),
     492            0 :                         addr: *pgb.get_peer_addr(),
     493            0 :                         conn_id: self.conn_id,
     494            0 :                         appname: self.appname.clone(),
     495            0 :                         feedback: ReplicationFeedback::Pageserver(PageserverFeedback::empty()),
     496            0 :                     },
     497            0 :                     interpreted_wal_reader: None,
     498            0 :                 }),
     499            0 :             )),
     500              :         };
     501              : 
     502              :         // Walsender can operate in one of two modes which we select by
     503              :         // application_name: give only committed WAL (used by pageserver) or all
     504              :         // existing WAL (up to flush_lsn, used by walproposer or peer recovery).
     505              :         // The second case is always driven by a consensus leader which term
     506              :         // must be supplied.
     507            0 :         let end_watch = if term.is_some() {
     508            0 :             EndWatch::Flush(tli.get_term_flush_lsn_watch_rx())
     509              :         } else {
     510            0 :             EndWatch::Commit(tli.get_commit_lsn_watch_rx())
     511              :         };
     512              :         // we don't check term here; it will be checked on first waiting/WAL reading anyway.
     513            0 :         let end_pos = end_watch.get();
     514            0 : 
     515            0 :         if end_pos < start_pos {
     516            0 :             warn!(
     517            0 :                 "requested start_pos {} is ahead of available WAL end_pos {}",
     518              :                 start_pos, end_pos
     519              :             );
     520            0 :         }
     521              : 
     522            0 :         info!(
     523            0 :             "starting streaming from {:?}, available WAL ends at {}, recovery={}, appname={:?}, protocol={:?}",
     524              :             start_pos,
     525              :             end_pos,
     526            0 :             matches!(end_watch, EndWatch::Flush(_)),
     527              :             appname,
     528            0 :             self.protocol(),
     529              :         );
     530              : 
     531              :         // switch to copy
     532            0 :         pgb.write_message(&BeMessage::CopyBothResponse).await?;
     533              : 
     534            0 :         let wal_reader = tli.get_walreader(start_pos).await?;
     535              : 
     536              :         // Split to concurrently receive and send data; replies are generally
     537              :         // not synchronized with sends, so this avoids deadlocks.
     538            0 :         let reader = pgb.split().context("START_REPLICATION split")?;
     539              : 
     540            0 :         let send_fut = match self.protocol() {
     541              :             PostgresClientProtocol::Vanilla => {
     542            0 :                 let sender = WalSender {
     543            0 :                     pgb,
     544            0 :                     // should succeed since we're already holding another guard
     545            0 :                     tli: tli.wal_residence_guard().await?,
     546            0 :                     appname: appname.clone(),
     547            0 :                     start_pos,
     548            0 :                     end_pos,
     549            0 :                     term,
     550            0 :                     end_watch,
     551            0 :                     ws_guard: ws_guard.clone(),
     552            0 :                     wal_reader,
     553            0 :                     send_buf: vec![0u8; MAX_SEND_SIZE],
     554            0 :                 };
     555            0 : 
     556            0 :                 FutureExt::boxed(sender.run())
     557              :             }
     558              :             PostgresClientProtocol::Interpreted {
     559            0 :                 format,
     560            0 :                 compression,
     561              :             } => {
     562            0 :                 let pg_version = tli.tli.get_state().await.1.server.pg_version / 10000;
     563            0 :                 let end_watch_view = end_watch.view();
     564            0 :                 let wal_residence_guard = tli.wal_residence_guard().await?;
     565            0 :                 let (tx, rx) = tokio::sync::mpsc::channel::<Batch>(2);
     566            0 :                 let shard = self.shard.unwrap();
     567            0 : 
     568            0 :                 if self.conf.wal_reader_fanout && !shard.is_unsharded() {
     569            0 :                     let ws_id = ws_guard.id();
     570            0 :                     ws_guard.walsenders().create_or_update_interpreted_reader(
     571            0 :                         ws_id,
     572            0 :                         start_pos,
     573            0 :                         self.conf.max_delta_for_fanout,
     574            0 :                         {
     575            0 :                             let tx = tx.clone();
     576            0 :                             |reader| {
     577            0 :                                 tracing::info!(
     578            0 :                                     "Fanning out interpreted wal reader at {}",
     579              :                                     start_pos
     580              :                                 );
     581            0 :                                 reader
     582            0 :                                     .fanout(shard, tx, start_pos)
     583            0 :                                     .with_context(|| "Failed to fan out reader")
     584            0 :                             }
     585            0 :                         },
     586            0 :                         || {
     587            0 :                             tracing::info!("Spawning interpreted wal reader at {}", start_pos);
     588              : 
     589            0 :                             let wal_stream = StreamingWalReader::new(
     590            0 :                                 wal_residence_guard,
     591            0 :                                 term,
     592            0 :                                 start_pos,
     593            0 :                                 end_pos,
     594            0 :                                 end_watch,
     595            0 :                                 MAX_SEND_SIZE,
     596            0 :                             );
     597            0 : 
     598            0 :                             InterpretedWalReader::spawn(
     599            0 :                                 wal_stream, start_pos, tx, shard, pg_version, &appname,
     600            0 :                             )
     601            0 :                         },
     602            0 :                     )?;
     603              : 
     604            0 :                     let sender = InterpretedWalSender {
     605            0 :                         format,
     606            0 :                         compression,
     607            0 :                         appname,
     608            0 :                         tli: tli.wal_residence_guard().await?,
     609            0 :                         start_lsn: start_pos,
     610            0 :                         pgb,
     611            0 :                         end_watch_view,
     612            0 :                         wal_sender_guard: ws_guard.clone(),
     613            0 :                         rx,
     614            0 :                     };
     615            0 : 
     616            0 :                     FutureExt::boxed(sender.run())
     617              :                 } else {
     618            0 :                     let wal_reader = StreamingWalReader::new(
     619            0 :                         wal_residence_guard,
     620            0 :                         term,
     621            0 :                         start_pos,
     622            0 :                         end_pos,
     623            0 :                         end_watch,
     624            0 :                         MAX_SEND_SIZE,
     625            0 :                     );
     626            0 : 
     627            0 :                     let reader = InterpretedWalReader::new(
     628            0 :                         wal_reader, start_pos, tx, shard, pg_version, None,
     629            0 :                     );
     630              : 
     631            0 :                     let sender = InterpretedWalSender {
     632            0 :                         format,
     633            0 :                         compression,
     634            0 :                         appname: appname.clone(),
     635            0 :                         tli: tli.wal_residence_guard().await?,
     636            0 :                         start_lsn: start_pos,
     637            0 :                         pgb,
     638            0 :                         end_watch_view,
     639            0 :                         wal_sender_guard: ws_guard.clone(),
     640            0 :                         rx,
     641            0 :                     };
     642            0 : 
     643            0 :                     FutureExt::boxed(async move {
     644              :                         // Sender returns an Err on all code paths.
     645              :                         // If the sender finishes first, we will drop the reader future.
     646              :                         // If the reader finishes first, the sender will finish too since
     647              :                         // the wal sender has dropped.
     648            0 :                         let res = tokio::try_join!(sender.run(), reader.run(start_pos, &appname));
     649            0 :                         match res.map(|_| ()) {
     650            0 :                             Ok(_) => unreachable!("sender finishes with Err by convention"),
     651            0 :                             err_res => err_res,
     652            0 :                         }
     653            0 :                     })
     654              :                 }
     655              :             }
     656              :         };
     657              : 
     658            0 :         let tli_cancel = tli.cancel.clone();
     659            0 : 
     660            0 :         let mut reply_reader = ReplyReader {
     661            0 :             reader,
     662            0 :             ws_guard: ws_guard.clone(),
     663            0 :             tli,
     664            0 :         };
     665              : 
     666            0 :         let res = tokio::select! {
     667              :             // todo: add read|write .context to these errors
     668            0 :             r = send_fut => r,
     669            0 :             r = reply_reader.run() => r,
     670            0 :             _ = tli_cancel.cancelled() => {
     671            0 :                 return Err(CopyStreamHandlerEnd::Cancelled);
     672              :             }
     673              :         };
     674              : 
     675            0 :         let ws_state = ws_guard
     676            0 :             .walsenders
     677            0 :             .mutex
     678            0 :             .lock()
     679            0 :             .get_slot(ws_guard.id)
     680            0 :             .clone();
     681            0 :         info!(
     682            0 :             "finished streaming to {}, feedback={:?}",
     683            0 :             ws_state.get_addr(),
     684            0 :             ws_state.get_feedback(),
     685              :         );
     686              : 
     687              :         // Join pg backend back.
     688            0 :         pgb.unsplit(reply_reader.reader)?;
     689              : 
     690            0 :         res
     691            0 :     }
     692              : }
     693              : 
     694              : /// TODO(vlad): maybe lift this instead
     695              : /// Walsender streams either up to commit_lsn (normally) or flush_lsn in the
     696              : /// given term (recovery by walproposer or peer safekeeper).
     697              : #[derive(Clone)]
     698              : pub(crate) enum EndWatch {
     699              :     Commit(Receiver<Lsn>),
     700              :     Flush(Receiver<TermLsn>),
     701              : }
     702              : 
     703              : impl EndWatch {
     704            0 :     pub(crate) fn view(&self) -> EndWatchView {
     705            0 :         EndWatchView(self.clone())
     706            0 :     }
     707              : 
     708              :     /// Get current end of WAL.
     709           12 :     pub(crate) fn get(&self) -> Lsn {
     710           12 :         match self {
     711           12 :             EndWatch::Commit(r) => *r.borrow(),
     712            0 :             EndWatch::Flush(r) => r.borrow().lsn,
     713              :         }
     714           12 :     }
     715              : 
     716              :     /// Wait for the update.
     717            7 :     pub(crate) async fn changed(&mut self) -> anyhow::Result<()> {
     718            7 :         match self {
     719            7 :             EndWatch::Commit(r) => r.changed().await?,
     720            0 :             EndWatch::Flush(r) => r.changed().await?,
     721              :         }
     722            3 :         Ok(())
     723            3 :     }
     724              : 
     725            4 :     pub(crate) async fn wait_for_lsn(
     726            4 :         &mut self,
     727            4 :         lsn: Lsn,
     728            4 :         client_term: Option<Term>,
     729            4 :     ) -> anyhow::Result<Lsn> {
     730              :         loop {
     731            7 :             let end_pos = self.get();
     732            7 :             if end_pos > lsn {
     733            0 :                 return Ok(end_pos);
     734            7 :             }
     735            7 :             if let EndWatch::Flush(rx) = &self {
     736            0 :                 let curr_term = rx.borrow().term;
     737            0 :                 if let Some(client_term) = client_term {
     738            0 :                     if curr_term != client_term {
     739            0 :                         bail!("term changed: requested {}, now {}", client_term, curr_term);
     740            0 :                     }
     741            0 :                 }
     742            7 :             }
     743            7 :             self.changed().await?;
     744              :         }
     745            0 :     }
     746              : }
     747              : 
     748              : pub(crate) struct EndWatchView(EndWatch);
     749              : 
     750              : impl EndWatchView {
     751            0 :     pub(crate) fn get(&self) -> Lsn {
     752            0 :         self.0.get()
     753            0 :     }
     754              : }
     755              : /// A half driving sending WAL.
     756              : struct WalSender<'a, IO> {
     757              :     pgb: &'a mut PostgresBackend<IO>,
     758              :     tli: WalResidentTimeline,
     759              :     appname: Option<String>,
     760              :     // Position since which we are sending next chunk.
     761              :     start_pos: Lsn,
     762              :     // WAL up to this position is known to be locally available.
     763              :     // Usually this is the same as the latest commit_lsn, but in case of
     764              :     // walproposer recovery, this is flush_lsn.
     765              :     //
     766              :     // We send this LSN to the receiver as wal_end, so that it knows how much
     767              :     // WAL this safekeeper has. This LSN should be as fresh as possible.
     768              :     end_pos: Lsn,
     769              :     /// When streaming uncommitted part, the term the client acts as the leader
     770              :     /// in. Streaming is stopped if local term changes to a different (higher)
     771              :     /// value.
     772              :     term: Option<Term>,
     773              :     /// Watch channel receiver to learn end of available WAL (and wait for its advancement).
     774              :     end_watch: EndWatch,
     775              :     ws_guard: Arc<WalSenderGuard>,
     776              :     wal_reader: WalReader,
     777              :     // buffer for readling WAL into to send it
     778              :     send_buf: Vec<u8>,
     779              : }
     780              : 
     781              : const POLL_STATE_TIMEOUT: Duration = Duration::from_secs(1);
     782              : 
     783              : impl<IO: AsyncRead + AsyncWrite + Unpin> WalSender<'_, IO> {
     784              :     /// Send WAL until
     785              :     /// - an error occurs
     786              :     /// - receiver is caughtup and there is no computes (if streaming up to commit_lsn)
     787              :     /// - timeline's cancellation token fires
     788              :     ///
     789              :     /// Err(CopyStreamHandlerEnd) is always returned; Result is used only for ?
     790              :     /// convenience.
     791            0 :     async fn run(mut self) -> Result<(), CopyStreamHandlerEnd> {
     792            0 :         let metric = WAL_READERS
     793            0 :             .get_metric_with_label_values(&[
     794            0 :                 "future",
     795            0 :                 self.appname.as_deref().unwrap_or("safekeeper"),
     796            0 :             ])
     797            0 :             .unwrap();
     798            0 : 
     799            0 :         metric.inc();
     800            0 :         scopeguard::defer! {
     801            0 :             metric.dec();
     802            0 :         }
     803              : 
     804              :         loop {
     805              :             // Wait for the next portion if it is not there yet, or just
     806              :             // update our end of WAL available for sending value, we
     807              :             // communicate it to the receiver.
     808            0 :             self.wait_wal().await?;
     809            0 :             assert!(
     810            0 :                 self.end_pos > self.start_pos,
     811            0 :                 "nothing to send after waiting for WAL"
     812              :             );
     813              : 
     814              :             // try to send as much as available, capped by MAX_SEND_SIZE
     815            0 :             let mut chunk_end_pos = self.start_pos + MAX_SEND_SIZE as u64;
     816            0 :             // if we went behind available WAL, back off
     817            0 :             if chunk_end_pos >= self.end_pos {
     818            0 :                 chunk_end_pos = self.end_pos;
     819            0 :             } else {
     820            0 :                 // If sending not up to end pos, round down to page boundary to
     821            0 :                 // avoid breaking WAL record not at page boundary, as protocol
     822            0 :                 // demands. See walsender.c (XLogSendPhysical).
     823            0 :                 chunk_end_pos = chunk_end_pos
     824            0 :                     .checked_sub(chunk_end_pos.block_offset())
     825            0 :                     .unwrap();
     826            0 :             }
     827            0 :             let send_size = (chunk_end_pos.0 - self.start_pos.0) as usize;
     828            0 :             let send_buf = &mut self.send_buf[..send_size];
     829              :             let send_size: usize;
     830              :             {
     831              :                 // If uncommitted part is being pulled, check that the term is
     832              :                 // still the expected one.
     833            0 :                 let _term_guard = if let Some(t) = self.term {
     834            0 :                     Some(self.tli.acquire_term(t).await?)
     835              :                 } else {
     836            0 :                     None
     837              :                 };
     838              :                 // Read WAL into buffer. send_size can be additionally capped to
     839              :                 // segment boundary here.
     840            0 :                 send_size = self.wal_reader.read(send_buf).await?
     841              :             };
     842            0 :             let send_buf = &send_buf[..send_size];
     843            0 : 
     844            0 :             // and send it, while respecting Timeline::cancel
     845            0 :             let msg = BeMessage::XLogData(XLogDataBody {
     846            0 :                 wal_start: self.start_pos.0,
     847            0 :                 wal_end: self.end_pos.0,
     848            0 :                 timestamp: get_current_timestamp(),
     849            0 :                 data: send_buf,
     850            0 :             });
     851            0 :             self.pgb.write_message(&msg).await?;
     852              : 
     853            0 :             if let Some(appname) = &self.appname {
     854            0 :                 if appname == "replica" {
     855            0 :                     failpoint_support::sleep_millis_async!("sk-send-wal-replica-sleep");
     856            0 :                 }
     857            0 :             }
     858            0 :             trace!(
     859            0 :                 "sent {} bytes of WAL {}-{}",
     860            0 :                 send_size,
     861            0 :                 self.start_pos,
     862            0 :                 self.start_pos + send_size as u64
     863              :             );
     864            0 :             self.start_pos += send_size as u64;
     865              :         }
     866            0 :     }
     867              : 
     868              :     /// wait until we have WAL to stream, sending keepalives and checking for
     869              :     /// exit in the meanwhile
     870            0 :     async fn wait_wal(&mut self) -> Result<(), CopyStreamHandlerEnd> {
     871              :         loop {
     872            0 :             self.end_pos = self.end_watch.get();
     873            0 :             let have_something_to_send = (|| {
     874            0 :                 fail::fail_point!(
     875            0 :                     "sk-pause-send",
     876            0 :                     self.appname.as_deref() != Some("pageserver"),
     877            0 :                     |_| { false }
     878            0 :                 );
     879            0 :                 self.end_pos > self.start_pos
     880            0 :             })();
     881            0 : 
     882            0 :             if have_something_to_send {
     883            0 :                 trace!("got end_pos {:?}, streaming", self.end_pos);
     884            0 :                 return Ok(());
     885            0 :             }
     886              : 
     887              :             // Wait for WAL to appear, now self.end_pos == self.start_pos.
     888            0 :             if let Some(lsn) = self.wait_for_lsn().await? {
     889            0 :                 self.end_pos = lsn;
     890            0 :                 trace!("got end_pos {:?}, streaming", self.end_pos);
     891            0 :                 return Ok(());
     892            0 :             }
     893            0 : 
     894            0 :             // Timed out waiting for WAL, check for termination and send KA.
     895            0 :             // Check for termination only if we are streaming up to commit_lsn
     896            0 :             // (to pageserver).
     897            0 :             if let EndWatch::Commit(_) = self.end_watch {
     898            0 :                 if let Some(remote_consistent_lsn) = self
     899            0 :                     .ws_guard
     900            0 :                     .walsenders
     901            0 :                     .get_ws_remote_consistent_lsn(self.ws_guard.id)
     902              :                 {
     903            0 :                     if self.tli.should_walsender_stop(remote_consistent_lsn).await {
     904              :                         // Terminate if there is nothing more to send.
     905              :                         // Note that "ending streaming" part of the string is used by
     906              :                         // pageserver to identify WalReceiverError::SuccessfulCompletion,
     907              :                         // do not change this string without updating pageserver.
     908            0 :                         return Err(CopyStreamHandlerEnd::ServerInitiated(format!(
     909            0 :                             "ending streaming to {:?} at {}, receiver is caughtup and there is no computes",
     910            0 :                             self.appname, self.start_pos,
     911            0 :                         )));
     912            0 :                     }
     913            0 :                 }
     914            0 :             }
     915              : 
     916            0 :             let msg = BeMessage::KeepAlive(WalSndKeepAlive {
     917            0 :                 wal_end: self.end_pos.0,
     918            0 :                 timestamp: get_current_timestamp(),
     919            0 :                 request_reply: true,
     920            0 :             });
     921            0 : 
     922            0 :             self.pgb.write_message(&msg).await?;
     923              :         }
     924            0 :     }
     925              : 
     926              :     /// Wait until we have available WAL > start_pos or timeout expires. Returns
     927              :     /// - Ok(Some(end_pos)) if needed lsn is successfully observed;
     928              :     /// - Ok(None) if timeout expired;
     929              :     /// - Err in case of error -- only if 1) term changed while fetching in recovery
     930              :     ///   mode 2) watch channel closed, which must never happen.
     931            0 :     async fn wait_for_lsn(&mut self) -> anyhow::Result<Option<Lsn>> {
     932            0 :         let fp = (|| {
     933            0 :             fail::fail_point!(
     934            0 :                 "sk-pause-send",
     935            0 :                 self.appname.as_deref() != Some("pageserver"),
     936            0 :                 |_| { true }
     937            0 :             );
     938            0 :             false
     939            0 :         })();
     940            0 :         if fp {
     941            0 :             tokio::time::sleep(POLL_STATE_TIMEOUT).await;
     942            0 :             return Ok(None);
     943            0 :         }
     944              : 
     945            0 :         let res = timeout(POLL_STATE_TIMEOUT, async move {
     946              :             loop {
     947            0 :                 let end_pos = self.end_watch.get();
     948            0 :                 if end_pos > self.start_pos {
     949            0 :                     return Ok(end_pos);
     950            0 :                 }
     951            0 :                 if let EndWatch::Flush(rx) = &self.end_watch {
     952            0 :                     let curr_term = rx.borrow().term;
     953            0 :                     if let Some(client_term) = self.term {
     954            0 :                         if curr_term != client_term {
     955            0 :                             bail!("term changed: requested {}, now {}", client_term, curr_term);
     956            0 :                         }
     957            0 :                     }
     958            0 :                 }
     959            0 :                 self.end_watch.changed().await?;
     960              :             }
     961            0 :         })
     962            0 :         .await;
     963              : 
     964            0 :         match res {
     965              :             // success
     966            0 :             Ok(Ok(commit_lsn)) => Ok(Some(commit_lsn)),
     967              :             // error inside closure
     968            0 :             Ok(Err(err)) => Err(err),
     969              :             // timeout
     970            0 :             Err(_) => Ok(None),
     971              :         }
     972            0 :     }
     973              : }
     974              : 
     975              : /// A half driving receiving replies.
     976              : struct ReplyReader<IO> {
     977              :     reader: PostgresBackendReader<IO>,
     978              :     ws_guard: Arc<WalSenderGuard>,
     979              :     tli: WalResidentTimeline,
     980              : }
     981              : 
     982              : impl<IO: AsyncRead + AsyncWrite + Unpin> ReplyReader<IO> {
     983            0 :     async fn run(&mut self) -> Result<(), CopyStreamHandlerEnd> {
     984              :         loop {
     985            0 :             let msg = self.reader.read_copy_message().await?;
     986            0 :             self.handle_feedback(&msg).await?
     987              :         }
     988            0 :     }
     989              : 
     990            0 :     async fn handle_feedback(&mut self, msg: &Bytes) -> anyhow::Result<()> {
     991            0 :         match msg.first().cloned() {
     992              :             Some(HOT_STANDBY_FEEDBACK_TAG_BYTE) => {
     993              :                 // Note: deserializing is on m[1..] because we skip the tag byte.
     994            0 :                 let mut hs_feedback = HotStandbyFeedback::des(&msg[1..])
     995            0 :                     .context("failed to deserialize HotStandbyFeedback")?;
     996              :                 // TODO: xmin/catalog_xmin are serialized by walreceiver.c in this way:
     997              :                 // pq_sendint32(&reply_message, xmin);
     998              :                 // pq_sendint32(&reply_message, xmin_epoch);
     999              :                 // So it is two big endian 32-bit words in low endian order!
    1000            0 :                 hs_feedback.xmin = hs_feedback.xmin.rotate_left(32);
    1001            0 :                 hs_feedback.catalog_xmin = hs_feedback.catalog_xmin.rotate_left(32);
    1002            0 :                 self.ws_guard
    1003            0 :                     .walsenders
    1004            0 :                     .record_hs_feedback(self.ws_guard.id, &hs_feedback);
    1005              :             }
    1006              :             Some(STANDBY_STATUS_UPDATE_TAG_BYTE) => {
    1007            0 :                 let reply =
    1008            0 :                     StandbyReply::des(&msg[1..]).context("failed to deserialize StandbyReply")?;
    1009            0 :                 self.ws_guard
    1010            0 :                     .walsenders
    1011            0 :                     .record_standby_reply(self.ws_guard.id, &reply);
    1012              :             }
    1013              :             Some(NEON_STATUS_UPDATE_TAG_BYTE) => {
    1014              :                 // pageserver sends this.
    1015              :                 // Note: deserializing is on m[9..] because we skip the tag byte and len bytes.
    1016            0 :                 let buf = Bytes::copy_from_slice(&msg[9..]);
    1017            0 :                 let ps_feedback = PageserverFeedback::parse(buf);
    1018            0 : 
    1019            0 :                 trace!("PageserverFeedback is {:?}", ps_feedback);
    1020            0 :                 self.ws_guard
    1021            0 :                     .walsenders
    1022            0 :                     .record_ps_feedback(self.ws_guard.id, &ps_feedback);
    1023            0 :                 self.tli
    1024            0 :                     .update_remote_consistent_lsn(ps_feedback.remote_consistent_lsn)
    1025            0 :                     .await;
    1026              :                 // in principle new remote_consistent_lsn could allow to
    1027              :                 // deactivate the timeline, but we check that regularly through
    1028              :                 // broker updated, not need to do it here
    1029              :             }
    1030            0 :             _ => warn!("unexpected message {:?}", msg),
    1031              :         }
    1032            0 :         Ok(())
    1033            0 :     }
    1034              : }
    1035              : 
    1036              : #[cfg(test)]
    1037              : mod tests {
    1038              :     use safekeeper_api::models::FullTransactionId;
    1039              :     use utils::id::{TenantId, TenantTimelineId, TimelineId};
    1040              : 
    1041              :     use super::*;
    1042              : 
    1043            4 :     fn mock_ttid() -> TenantTimelineId {
    1044            4 :         TenantTimelineId {
    1045            4 :             tenant_id: TenantId::from_slice(&[0x00; 16]).unwrap(),
    1046            4 :             timeline_id: TimelineId::from_slice(&[0x00; 16]).unwrap(),
    1047            4 :         }
    1048            4 :     }
    1049              : 
    1050            4 :     fn mock_addr() -> SocketAddr {
    1051            4 :         "127.0.0.1:8080".parse().unwrap()
    1052            4 :     }
    1053              : 
    1054              :     // add to wss specified feedback setting other fields to dummy values
    1055            4 :     fn push_feedback(wss: &mut WalSendersShared, feedback: ReplicationFeedback) {
    1056            4 :         let walsender_state = WalSenderState::Vanilla(VanillaWalSenderInternalState {
    1057            4 :             ttid: mock_ttid(),
    1058            4 :             addr: mock_addr(),
    1059            4 :             conn_id: 1,
    1060            4 :             appname: None,
    1061            4 :             feedback,
    1062            4 :         });
    1063            4 :         wss.slots.push(Some(walsender_state))
    1064            4 :     }
    1065              : 
    1066              :     // form standby feedback with given hot standby feedback ts/xmin and the
    1067              :     // rest set to dummy values.
    1068            4 :     fn hs_feedback(ts: TimestampTz, xmin: FullTransactionId) -> ReplicationFeedback {
    1069            4 :         ReplicationFeedback::Standby(StandbyFeedback {
    1070            4 :             reply: StandbyReply::empty(),
    1071            4 :             hs_feedback: HotStandbyFeedback {
    1072            4 :                 ts,
    1073            4 :                 xmin,
    1074            4 :                 catalog_xmin: 0,
    1075            4 :             },
    1076            4 :         })
    1077            4 :     }
    1078              : 
    1079              :     // test that hs aggregation works as expected
    1080              :     #[test]
    1081            1 :     fn test_hs_feedback_no_valid() {
    1082            1 :         let mut wss = WalSendersShared::new();
    1083            1 :         push_feedback(&mut wss, hs_feedback(1, INVALID_FULL_TRANSACTION_ID));
    1084            1 :         wss.update_reply_feedback();
    1085            1 :         assert_eq!(
    1086            1 :             wss.agg_standby_feedback.hs_feedback.xmin,
    1087            1 :             INVALID_FULL_TRANSACTION_ID
    1088            1 :         );
    1089            1 :     }
    1090              : 
    1091              :     #[test]
    1092            1 :     fn test_hs_feedback() {
    1093            1 :         let mut wss = WalSendersShared::new();
    1094            1 :         push_feedback(&mut wss, hs_feedback(1, INVALID_FULL_TRANSACTION_ID));
    1095            1 :         push_feedback(&mut wss, hs_feedback(1, 42));
    1096            1 :         push_feedback(&mut wss, hs_feedback(1, 64));
    1097            1 :         wss.update_reply_feedback();
    1098            1 :         assert_eq!(wss.agg_standby_feedback.hs_feedback.xmin, 42);
    1099            1 :     }
    1100              : }
        

Generated by: LCOV version 2.1-beta