LCOV - code coverage report
Current view: top level - pageserver/src/tenant/timeline/walreceiver - walreceiver_connection.rs (source / functions) Coverage Total Hit
Test: b9d67f908f91f00e353a27440ba89f642a869959.info Lines: 0.0 % 359 0
Test Date: 2024-11-19 21:44:13 Functions: 0.0 % 20 0

            Line data    Source code
       1              : //! Actual Postgres connection handler to stream WAL to the server.
       2              : 
       3              : use std::{
       4              :     error::Error,
       5              :     pin::pin,
       6              :     str::FromStr,
       7              :     sync::Arc,
       8              :     time::{Duration, SystemTime},
       9              : };
      10              : 
      11              : use anyhow::{anyhow, Context};
      12              : use bytes::BytesMut;
      13              : use chrono::{NaiveDateTime, Utc};
      14              : use fail::fail_point;
      15              : use futures::StreamExt;
      16              : use postgres::{error::SqlState, SimpleQueryMessage, SimpleQueryRow};
      17              : use postgres_ffi::WAL_SEGMENT_SIZE;
      18              : use postgres_ffi::{v14::xlog_utils::normalize_lsn, waldecoder::WalDecodeError};
      19              : use postgres_protocol::message::backend::ReplicationMessage;
      20              : use postgres_types::PgLsn;
      21              : use tokio::{select, sync::watch, time};
      22              : use tokio_postgres::{replication::ReplicationStream, Client};
      23              : use tokio_util::sync::CancellationToken;
      24              : use tracing::{debug, error, info, trace, warn, Instrument};
      25              : use wal_decoder::models::{FlushUncommittedRecords, InterpretedWalRecord};
      26              : 
      27              : use super::TaskStateUpdate;
      28              : use crate::{
      29              :     context::RequestContext,
      30              :     metrics::{LIVE_CONNECTIONS, WALRECEIVER_STARTED_CONNECTIONS, WAL_INGEST},
      31              :     pgdatadir_mapping::DatadirModification,
      32              :     task_mgr::{TaskKind, WALRECEIVER_RUNTIME},
      33              :     tenant::{debug_assert_current_span_has_tenant_and_timeline_id, Timeline, WalReceiverInfo},
      34              :     walingest::WalIngest,
      35              : };
      36              : use postgres_backend::is_expected_io_error;
      37              : use postgres_connection::PgConnectionConfig;
      38              : use postgres_ffi::waldecoder::WalStreamDecoder;
      39              : use utils::{id::NodeId, lsn::Lsn};
      40              : use utils::{pageserver_feedback::PageserverFeedback, sync::gate::GateError};
      41              : 
      42              : /// Status of the connection.
      43              : #[derive(Debug, Clone, Copy)]
      44              : pub(super) struct WalConnectionStatus {
      45              :     /// If we were able to initiate a postgres connection, this means that safekeeper process is at least running.
      46              :     pub is_connected: bool,
      47              :     /// Defines a healthy connection as one on which pageserver received WAL from safekeeper
      48              :     /// and is able to process it in walingest without errors.
      49              :     pub has_processed_wal: bool,
      50              :     /// Connection establishment time or the timestamp of a latest connection message received.
      51              :     pub latest_connection_update: NaiveDateTime,
      52              :     /// Time of the latest WAL message received.
      53              :     pub latest_wal_update: NaiveDateTime,
      54              :     /// Latest WAL update contained WAL up to this LSN. Next WAL message with start from that LSN.
      55              :     pub streaming_lsn: Option<Lsn>,
      56              :     /// Latest commit_lsn received from the safekeeper. Can be zero if no message has been received yet.
      57              :     pub commit_lsn: Option<Lsn>,
      58              :     /// The node it is connected to
      59              :     pub node: NodeId,
      60              : }
      61              : 
      62              : pub(super) enum WalReceiverError {
      63              :     /// An error of a type that does not indicate an issue, e.g. a connection closing
      64              :     ExpectedSafekeeperError(postgres::Error),
      65              :     /// An "error" message that carries a SUCCESSFUL_COMPLETION status code.  Carries
      66              :     /// the message part of the original postgres error
      67              :     SuccessfulCompletion(String),
      68              :     /// Generic error
      69              :     Other(anyhow::Error),
      70              :     ClosedGate,
      71              : }
      72              : 
      73              : impl From<tokio_postgres::Error> for WalReceiverError {
      74            0 :     fn from(err: tokio_postgres::Error) -> Self {
      75            0 :         if let Some(dberror) = err.as_db_error().filter(|db_error| {
      76            0 :             db_error.code() == &SqlState::SUCCESSFUL_COMPLETION
      77            0 :                 && db_error.message().contains("ending streaming")
      78            0 :         }) {
      79              :             // Strip the outer DbError, which carries a misleading "error" severity
      80            0 :             Self::SuccessfulCompletion(dberror.message().to_string())
      81            0 :         } else if err.is_closed()
      82            0 :             || err
      83            0 :                 .source()
      84            0 :                 .and_then(|source| source.downcast_ref::<std::io::Error>())
      85            0 :                 .map(is_expected_io_error)
      86            0 :                 .unwrap_or(false)
      87              :         {
      88            0 :             Self::ExpectedSafekeeperError(err)
      89              :         } else {
      90            0 :             Self::Other(anyhow::Error::new(err))
      91              :         }
      92            0 :     }
      93              : }
      94              : 
      95              : impl From<anyhow::Error> for WalReceiverError {
      96            0 :     fn from(err: anyhow::Error) -> Self {
      97            0 :         Self::Other(err)
      98            0 :     }
      99              : }
     100              : 
     101              : impl From<WalDecodeError> for WalReceiverError {
     102            0 :     fn from(err: WalDecodeError) -> Self {
     103            0 :         Self::Other(anyhow::Error::new(err))
     104            0 :     }
     105              : }
     106              : 
     107              : /// Open a connection to the given safekeeper and receive WAL, sending back progress
     108              : /// messages as we go.
     109              : #[allow(clippy::too_many_arguments)]
     110            0 : pub(super) async fn handle_walreceiver_connection(
     111            0 :     timeline: Arc<Timeline>,
     112            0 :     wal_source_connconf: PgConnectionConfig,
     113            0 :     events_sender: watch::Sender<TaskStateUpdate<WalConnectionStatus>>,
     114            0 :     cancellation: CancellationToken,
     115            0 :     connect_timeout: Duration,
     116            0 :     ctx: RequestContext,
     117            0 :     node: NodeId,
     118            0 :     ingest_batch_size: u64,
     119            0 : ) -> Result<(), WalReceiverError> {
     120            0 :     debug_assert_current_span_has_tenant_and_timeline_id();
     121              : 
     122              :     // prevent timeline shutdown from finishing until we have exited
     123            0 :     let _guard = timeline.gate.enter().map_err(|e| match e {
     124            0 :         GateError::GateClosed => WalReceiverError::ClosedGate,
     125            0 :     })?;
     126              :     // This function spawns a side-car task (WalReceiverConnectionPoller).
     127              :     // Get its gate guard now as well.
     128            0 :     let poller_guard = timeline.gate.enter().map_err(|e| match e {
     129            0 :         GateError::GateClosed => WalReceiverError::ClosedGate,
     130            0 :     })?;
     131              : 
     132            0 :     WALRECEIVER_STARTED_CONNECTIONS.inc();
     133            0 : 
     134            0 :     // Connect to the database in replication mode.
     135            0 :     info!("connecting to {wal_source_connconf:?}");
     136              : 
     137            0 :     let (replication_client, connection) = {
     138            0 :         let mut config = wal_source_connconf.to_tokio_postgres_config();
     139            0 :         config.application_name("pageserver");
     140            0 :         config.replication_mode(tokio_postgres::config::ReplicationMode::Physical);
     141            0 :         match time::timeout(connect_timeout, config.connect(postgres::NoTls)).await {
     142            0 :             Ok(client_and_conn) => client_and_conn?,
     143            0 :             Err(_elapsed) => {
     144            0 :                 // Timing out to connect to a safekeeper node could happen long time, due to
     145            0 :                 // many reasons that pageserver cannot control.
     146            0 :                 // Do not produce an error, but make it visible, that timeouts happen by logging the `event.
     147            0 :                 info!("Timed out while waiting {connect_timeout:?} for walreceiver connection to open");
     148            0 :                 return Ok(());
     149              :             }
     150              :         }
     151              :     };
     152              : 
     153            0 :     debug!("connected!");
     154            0 :     let mut connection_status = WalConnectionStatus {
     155            0 :         is_connected: true,
     156            0 :         has_processed_wal: false,
     157            0 :         latest_connection_update: Utc::now().naive_utc(),
     158            0 :         latest_wal_update: Utc::now().naive_utc(),
     159            0 :         streaming_lsn: None,
     160            0 :         commit_lsn: None,
     161            0 :         node,
     162            0 :     };
     163            0 :     if let Err(e) = events_sender.send(TaskStateUpdate::Progress(connection_status)) {
     164            0 :         warn!("Wal connection event listener dropped right after connection init, aborting the connection: {e}");
     165            0 :         return Ok(());
     166            0 :     }
     167            0 : 
     168            0 :     // The connection object performs the actual communication with the database,
     169            0 :     // so spawn it off to run on its own. It shouldn't outlive this function, but,
     170            0 :     // due to lack of async drop, we can't enforce that. However, we ensure that
     171            0 :     // 1. it is sensitive to `cancellation` and
     172            0 :     // 2. holds the Timeline gate open so that after timeline shutdown,
     173            0 :     //    we know this task is gone.
     174            0 :     let _connection_ctx = ctx.detached_child(
     175            0 :         TaskKind::WalReceiverConnectionPoller,
     176            0 :         ctx.download_behavior(),
     177            0 :     );
     178            0 :     let connection_cancellation = cancellation.clone();
     179            0 :     WALRECEIVER_RUNTIME.spawn(
     180            0 :         async move {
     181            0 :             debug_assert_current_span_has_tenant_and_timeline_id();
     182            0 :             select! {
     183            0 :                 connection_result = connection => match connection_result {
     184            0 :                     Ok(()) => debug!("Walreceiver db connection closed"),
     185            0 :                     Err(connection_error) => {
     186            0 :                         match WalReceiverError::from(connection_error) {
     187            0 :                             WalReceiverError::ExpectedSafekeeperError(_) => {
     188            0 :                                 // silence, because most likely we've already exited the outer call
     189            0 :                                 // with a similar error.
     190            0 :                             },
     191            0 :                             WalReceiverError::SuccessfulCompletion(_) => {}
     192            0 :                             WalReceiverError::ClosedGate => {
     193            0 :                                 // doesn't happen at runtime
     194            0 :                             }
     195            0 :                             WalReceiverError::Other(err) => {
     196            0 :                                 warn!("Connection aborted: {err:#}")
     197              :                             }
     198              :                         }
     199              :                     }
     200              :                 },
     201            0 :                 _ = connection_cancellation.cancelled() => debug!("Connection cancelled"),
     202              :             }
     203            0 :             drop(poller_guard);
     204            0 :         }
     205              :         // Enrich the log lines emitted by this closure with meaningful context.
     206              :         // TODO: technically, this task outlives the surrounding function, so, the
     207              :         // spans won't be properly nested.
     208            0 :         .instrument(tracing::info_span!("poller")),
     209              :     );
     210              : 
     211            0 :     let _guard = LIVE_CONNECTIONS
     212            0 :         .with_label_values(&["wal_receiver"])
     213            0 :         .guard();
     214              : 
     215            0 :     let identify = identify_system(&replication_client).await?;
     216            0 :     info!("{identify:?}");
     217              : 
     218            0 :     let end_of_wal = Lsn::from(u64::from(identify.xlogpos));
     219            0 :     let mut caught_up = false;
     220            0 : 
     221            0 :     connection_status.latest_connection_update = Utc::now().naive_utc();
     222            0 :     connection_status.latest_wal_update = Utc::now().naive_utc();
     223            0 :     connection_status.commit_lsn = Some(end_of_wal);
     224            0 :     if let Err(e) = events_sender.send(TaskStateUpdate::Progress(connection_status)) {
     225            0 :         warn!("Wal connection event listener dropped after IDENTIFY_SYSTEM, aborting the connection: {e}");
     226            0 :         return Ok(());
     227            0 :     }
     228            0 : 
     229            0 :     //
     230            0 :     // Start streaming the WAL, from where we left off previously.
     231            0 :     //
     232            0 :     // If we had previously received WAL up to some point in the middle of a WAL record, we
     233            0 :     // better start from the end of last full WAL record, not in the middle of one.
     234            0 :     let mut last_rec_lsn = timeline.get_last_record_lsn();
     235            0 :     let mut startpoint = last_rec_lsn;
     236            0 : 
     237            0 :     if startpoint == Lsn(0) {
     238            0 :         return Err(WalReceiverError::Other(anyhow!("No previous WAL position")));
     239            0 :     }
     240            0 : 
     241            0 :     // There might be some padding after the last full record, skip it.
     242            0 :     startpoint += startpoint.calc_padding(8u32);
     243            0 : 
     244            0 :     // If the starting point is at a WAL page boundary, skip past the page header. We don't need the page headers
     245            0 :     // for anything, and in some corner cases, the compute node might have never generated the WAL for page headers
     246            0 :     //. That happens if you create a branch at page boundary: the start point of the branch is at the page boundary,
     247            0 :     // but when the compute node first starts on the branch, we normalize the first REDO position to just after the page
     248            0 :     // header (see generate_pg_control()), so the WAL for the page header is never streamed from the compute node
     249            0 :     //  to the safekeepers.
     250            0 :     startpoint = normalize_lsn(startpoint, WAL_SEGMENT_SIZE);
     251            0 : 
     252            0 :     info!("last_record_lsn {last_rec_lsn} starting replication from {startpoint}, safekeeper is at {end_of_wal}...");
     253              : 
     254            0 :     let query = format!("START_REPLICATION PHYSICAL {startpoint}");
     255              : 
     256            0 :     let copy_stream = replication_client.copy_both_simple(&query).await?;
     257            0 :     let mut physical_stream = pin!(ReplicationStream::new(copy_stream));
     258            0 : 
     259            0 :     let mut waldecoder = WalStreamDecoder::new(startpoint, timeline.pg_version);
     260              : 
     261            0 :     let mut walingest = WalIngest::new(timeline.as_ref(), startpoint, &ctx).await?;
     262              : 
     263            0 :     while let Some(replication_message) = {
     264            0 :         select! {
     265            0 :             _ = cancellation.cancelled() => {
     266            0 :                 debug!("walreceiver interrupted");
     267            0 :                 None
     268              :             }
     269            0 :             replication_message = physical_stream.next() => replication_message,
     270              :         }
     271              :     } {
     272            0 :         let replication_message = replication_message?;
     273              : 
     274            0 :         let now = Utc::now().naive_utc();
     275            0 :         let last_rec_lsn_before_msg = last_rec_lsn;
     276            0 : 
     277            0 :         // Update the connection status before processing the message. If the message processing
     278            0 :         // fails (e.g. in walingest), we still want to know latests LSNs from the safekeeper.
     279            0 :         match &replication_message {
     280            0 :             ReplicationMessage::XLogData(xlog_data) => {
     281            0 :                 connection_status.latest_connection_update = now;
     282            0 :                 connection_status.commit_lsn = Some(Lsn::from(xlog_data.wal_end()));
     283            0 :                 connection_status.streaming_lsn = Some(Lsn::from(
     284            0 :                     xlog_data.wal_start() + xlog_data.data().len() as u64,
     285            0 :                 ));
     286            0 :                 if !xlog_data.data().is_empty() {
     287            0 :                     connection_status.latest_wal_update = now;
     288            0 :                 }
     289              :             }
     290            0 :             ReplicationMessage::PrimaryKeepAlive(keepalive) => {
     291            0 :                 connection_status.latest_connection_update = now;
     292            0 :                 connection_status.commit_lsn = Some(Lsn::from(keepalive.wal_end()));
     293            0 :             }
     294            0 :             &_ => {}
     295              :         };
     296            0 :         if let Err(e) = events_sender.send(TaskStateUpdate::Progress(connection_status)) {
     297            0 :             warn!("Wal connection event listener dropped, aborting the connection: {e}");
     298            0 :             return Ok(());
     299            0 :         }
     300              : 
     301            0 :         let status_update = match replication_message {
     302            0 :             ReplicationMessage::XLogData(xlog_data) => {
     303            0 :                 // Pass the WAL data to the decoder, and see if we can decode
     304            0 :                 // more records as a result.
     305            0 :                 let data = xlog_data.data();
     306            0 :                 let startlsn = Lsn::from(xlog_data.wal_start());
     307            0 :                 let endlsn = startlsn + data.len() as u64;
     308            0 : 
     309            0 :                 trace!("received XLogData between {startlsn} and {endlsn}");
     310              : 
     311            0 :                 WAL_INGEST.bytes_received.inc_by(data.len() as u64);
     312            0 :                 waldecoder.feed_bytes(data);
     313            0 : 
     314            0 :                 {
     315            0 :                     let mut modification = timeline.begin_modification(startlsn);
     316            0 :                     let mut uncommitted_records = 0;
     317            0 :                     let mut filtered_records = 0;
     318              : 
     319            0 :                     async fn commit(
     320            0 :                         modification: &mut DatadirModification<'_>,
     321            0 :                         uncommitted: &mut u64,
     322            0 :                         filtered: &mut u64,
     323            0 :                         ctx: &RequestContext,
     324            0 :                     ) -> anyhow::Result<()> {
     325            0 :                         WAL_INGEST
     326            0 :                             .records_committed
     327            0 :                             .inc_by(*uncommitted - *filtered);
     328            0 :                         modification.commit(ctx).await?;
     329            0 :                         *uncommitted = 0;
     330            0 :                         *filtered = 0;
     331            0 :                         Ok(())
     332            0 :                     }
     333              : 
     334            0 :                     while let Some((next_record_lsn, recdata)) = waldecoder.poll_decode()? {
     335              :                         // It is important to deal with the aligned records as lsn in getPage@LSN is
     336              :                         // aligned and can be several bytes bigger. Without this alignment we are
     337              :                         // at risk of hitting a deadlock.
     338            0 :                         if !next_record_lsn.is_aligned() {
     339            0 :                             return Err(WalReceiverError::Other(anyhow!("LSN not aligned")));
     340            0 :                         }
     341              : 
     342              :                         // Deserialize and interpret WAL record
     343            0 :                         let interpreted = InterpretedWalRecord::from_bytes_filtered(
     344            0 :                             recdata,
     345            0 :                             modification.tline.get_shard_identity(),
     346            0 :                             next_record_lsn,
     347            0 :                             modification.tline.pg_version,
     348            0 :                         )?;
     349              : 
     350            0 :                         if matches!(interpreted.flush_uncommitted, FlushUncommittedRecords::Yes)
     351            0 :                             && uncommitted_records > 0
     352              :                         {
     353              :                             // Special case: legacy PG database creations operate by reading pages from a 'template' database:
     354              :                             // these are the only kinds of WAL record that require reading data blocks while ingesting.  Ensure
     355              :                             // all earlier writes of data blocks are visible by committing any modification in flight.
     356            0 :                             commit(
     357            0 :                                 &mut modification,
     358            0 :                                 &mut uncommitted_records,
     359            0 :                                 &mut filtered_records,
     360            0 :                                 &ctx,
     361            0 :                             )
     362            0 :                             .await?;
     363            0 :                         }
     364              : 
     365              :                         // Ingest the records without immediately committing them.
     366            0 :                         let ingested = walingest
     367            0 :                             .ingest_record(interpreted, &mut modification, &ctx)
     368            0 :                             .await
     369            0 :                             .with_context(|| {
     370            0 :                                 format!("could not ingest record at {next_record_lsn}")
     371            0 :                             })?;
     372            0 :                         if !ingested {
     373            0 :                             tracing::debug!("ingest: filtered out record @ LSN {next_record_lsn}");
     374            0 :                             WAL_INGEST.records_filtered.inc();
     375            0 :                             filtered_records += 1;
     376            0 :                         }
     377              : 
     378              :                         // FIXME: this cannot be made pausable_failpoint without fixing the
     379              :                         // failpoint library; in tests, the added amount of debugging will cause us
     380              :                         // to timeout the tests.
     381            0 :                         fail_point!("walreceiver-after-ingest");
     382            0 : 
     383            0 :                         last_rec_lsn = next_record_lsn;
     384            0 : 
     385            0 :                         // Commit every ingest_batch_size records. Even if we filtered out
     386            0 :                         // all records, we still need to call commit to advance the LSN.
     387            0 :                         uncommitted_records += 1;
     388            0 :                         if uncommitted_records >= ingest_batch_size
     389            0 :                             || modification.approx_pending_bytes()
     390            0 :                                 > DatadirModification::MAX_PENDING_BYTES
     391              :                         {
     392            0 :                             commit(
     393            0 :                                 &mut modification,
     394            0 :                                 &mut uncommitted_records,
     395            0 :                                 &mut filtered_records,
     396            0 :                                 &ctx,
     397            0 :                             )
     398            0 :                             .await?;
     399            0 :                         }
     400              :                     }
     401              : 
     402              :                     // Commit the remaining records.
     403            0 :                     if uncommitted_records > 0 {
     404            0 :                         commit(
     405            0 :                             &mut modification,
     406            0 :                             &mut uncommitted_records,
     407            0 :                             &mut filtered_records,
     408            0 :                             &ctx,
     409            0 :                         )
     410            0 :                         .await?;
     411            0 :                     }
     412              :                 }
     413              : 
     414            0 :                 if !caught_up && endlsn >= end_of_wal {
     415            0 :                     info!("caught up at LSN {endlsn}");
     416            0 :                     caught_up = true;
     417            0 :                 }
     418              : 
     419            0 :                 Some(endlsn)
     420              :             }
     421              : 
     422            0 :             ReplicationMessage::PrimaryKeepAlive(keepalive) => {
     423            0 :                 let wal_end = keepalive.wal_end();
     424            0 :                 let timestamp = keepalive.timestamp();
     425            0 :                 let reply_requested = keepalive.reply() != 0;
     426            0 : 
     427            0 :                 trace!("received PrimaryKeepAlive(wal_end: {wal_end}, timestamp: {timestamp:?} reply: {reply_requested})");
     428              : 
     429            0 :                 if reply_requested {
     430            0 :                     Some(last_rec_lsn)
     431              :                 } else {
     432            0 :                     None
     433              :                 }
     434              :             }
     435              : 
     436            0 :             _ => None,
     437              :         };
     438              : 
     439            0 :         if !connection_status.has_processed_wal && last_rec_lsn > last_rec_lsn_before_msg {
     440              :             // We have successfully processed at least one WAL record.
     441            0 :             connection_status.has_processed_wal = true;
     442            0 :             if let Err(e) = events_sender.send(TaskStateUpdate::Progress(connection_status)) {
     443            0 :                 warn!("Wal connection event listener dropped, aborting the connection: {e}");
     444            0 :                 return Ok(());
     445            0 :             }
     446            0 :         }
     447              : 
     448            0 :         if let Some(last_lsn) = status_update {
     449            0 :             let timeline_remote_consistent_lsn = timeline
     450            0 :                 .get_remote_consistent_lsn_visible()
     451            0 :                 .unwrap_or(Lsn(0));
     452            0 : 
     453            0 :             // The last LSN we processed. It is not guaranteed to survive pageserver crash.
     454            0 :             let last_received_lsn = last_lsn;
     455            0 :             // `disk_consistent_lsn` is the LSN at which page server guarantees local persistence of all received data
     456            0 :             let disk_consistent_lsn = timeline.get_disk_consistent_lsn();
     457            0 :             // The last LSN that is synced to remote storage and is guaranteed to survive pageserver crash
     458            0 :             // Used by safekeepers to remove WAL preceding `remote_consistent_lsn`.
     459            0 :             let remote_consistent_lsn = timeline_remote_consistent_lsn;
     460            0 :             let ts = SystemTime::now();
     461            0 : 
     462            0 :             // Update the status about what we just received. This is shown in the mgmt API.
     463            0 :             let last_received_wal = WalReceiverInfo {
     464            0 :                 wal_source_connconf: wal_source_connconf.clone(),
     465            0 :                 last_received_msg_lsn: last_lsn,
     466            0 :                 last_received_msg_ts: ts
     467            0 :                     .duration_since(SystemTime::UNIX_EPOCH)
     468            0 :                     .expect("Received message time should be before UNIX EPOCH!")
     469            0 :                     .as_micros(),
     470            0 :             };
     471            0 :             *timeline.last_received_wal.lock().unwrap() = Some(last_received_wal);
     472              : 
     473              :             // Send the replication feedback message.
     474              :             // Regular standby_status_update fields are put into this message.
     475            0 :             let current_timeline_size = if timeline.tenant_shard_id.is_shard_zero() {
     476            0 :                 timeline
     477            0 :                     .get_current_logical_size(
     478            0 :                         crate::tenant::timeline::GetLogicalSizePriority::User,
     479            0 :                         &ctx,
     480            0 :                     )
     481            0 :                     // FIXME: https://github.com/neondatabase/neon/issues/5963
     482            0 :                     .size_dont_care_about_accuracy()
     483              :             } else {
     484              :                 // Non-zero shards send zero for logical size.  The safekeeper will ignore
     485              :                 // this number.  This is because in a sharded tenant, only shard zero maintains
     486              :                 // accurate logical size.
     487            0 :                 0
     488              :             };
     489              : 
     490            0 :             let status_update = PageserverFeedback {
     491            0 :                 current_timeline_size,
     492            0 :                 last_received_lsn,
     493            0 :                 disk_consistent_lsn,
     494            0 :                 remote_consistent_lsn,
     495            0 :                 replytime: ts,
     496            0 :                 shard_number: timeline.tenant_shard_id.shard_number.0 as u32,
     497            0 :             };
     498            0 : 
     499            0 :             debug!("neon_status_update {status_update:?}");
     500              : 
     501            0 :             let mut data = BytesMut::new();
     502            0 :             status_update.serialize(&mut data);
     503            0 :             physical_stream
     504            0 :                 .as_mut()
     505            0 :                 .zenith_status_update(data.len() as u64, &data)
     506            0 :                 .await?;
     507            0 :         }
     508              :     }
     509              : 
     510            0 :     Ok(())
     511            0 : }
     512              : 
     513              : /// Data returned from the postgres `IDENTIFY_SYSTEM` command
     514              : ///
     515              : /// See the [postgres docs] for more details.
     516              : ///
     517              : /// [postgres docs]: https://www.postgresql.org/docs/current/protocol-replication.html
     518              : #[derive(Debug)]
     519              : // As of nightly 2021-09-11, fields that are only read by the type's `Debug` impl still count as
     520              : // unused. Relevant issue: https://github.com/rust-lang/rust/issues/88900
     521              : #[allow(dead_code)]
     522              : struct IdentifySystem {
     523              :     systemid: u64,
     524              :     timeline: u32,
     525              :     xlogpos: PgLsn,
     526              :     dbname: Option<String>,
     527              : }
     528              : 
     529              : /// There was a problem parsing the response to
     530              : /// a postgres IDENTIFY_SYSTEM command.
     531            0 : #[derive(Debug, thiserror::Error)]
     532              : #[error("IDENTIFY_SYSTEM parse error")]
     533              : struct IdentifyError;
     534              : 
     535              : /// Run the postgres `IDENTIFY_SYSTEM` command
     536            0 : async fn identify_system(client: &Client) -> anyhow::Result<IdentifySystem> {
     537            0 :     let query_str = "IDENTIFY_SYSTEM";
     538            0 :     let response = client.simple_query(query_str).await?;
     539              : 
     540              :     // get(N) from row, then parse it as some destination type.
     541            0 :     fn get_parse<T>(row: &SimpleQueryRow, idx: usize) -> Result<T, IdentifyError>
     542            0 :     where
     543            0 :         T: FromStr,
     544            0 :     {
     545            0 :         let val = row.get(idx).ok_or(IdentifyError)?;
     546            0 :         val.parse::<T>().or(Err(IdentifyError))
     547            0 :     }
     548              : 
     549              :     // extract the row contents into an IdentifySystem struct.
     550              :     // written as a closure so I can use ? for Option here.
     551            0 :     if let Some(SimpleQueryMessage::Row(first_row)) = response.first() {
     552              :         Ok(IdentifySystem {
     553            0 :             systemid: get_parse(first_row, 0)?,
     554            0 :             timeline: get_parse(first_row, 1)?,
     555            0 :             xlogpos: get_parse(first_row, 2)?,
     556            0 :             dbname: get_parse(first_row, 3).ok(),
     557              :         })
     558              :     } else {
     559            0 :         Err(IdentifyError.into())
     560              :     }
     561            0 : }
        

Generated by: LCOV version 2.1-beta