LCOV - code coverage report
Current view: top level - pageserver/src - basebackup.rs (source / functions) Coverage Total Hit
Test: 07bee600374ccd486c69370d0972d9035964fe68.info Lines: 0.0 % 510 0
Test Date: 2025-02-20 13:11:02 Functions: 0.0 % 176 0

            Line data    Source code
       1              : //!
       2              : //! Generate a tarball with files needed to bootstrap ComputeNode.
       3              : //!
       4              : //! TODO: this module has nothing to do with PostgreSQL pg_basebackup.
       5              : //! It could use a better name.
       6              : //!
       7              : //! Stateless Postgres compute node is launched by sending a tarball
       8              : //! which contains non-relational data (multixacts, clog, filenodemaps, twophase files),
       9              : //! generated pg_control and dummy segment of WAL.
      10              : //! This module is responsible for creation of such tarball
      11              : //! from data stored in object storage.
      12              : //!
      13              : use anyhow::{anyhow, Context};
      14              : use bytes::{BufMut, Bytes, BytesMut};
      15              : use fail::fail_point;
      16              : use pageserver_api::key::{rel_block_to_key, Key};
      17              : use postgres_ffi::pg_constants;
      18              : use std::fmt::Write as FmtWrite;
      19              : use std::time::{Instant, SystemTime};
      20              : use tokio::io;
      21              : use tokio::io::AsyncWrite;
      22              : use tracing::*;
      23              : 
      24              : use tokio_tar::{Builder, EntryType, Header};
      25              : 
      26              : use crate::context::RequestContext;
      27              : use crate::pgdatadir_mapping::Version;
      28              : use crate::tenant::storage_layer::IoConcurrency;
      29              : use crate::tenant::Timeline;
      30              : use pageserver_api::reltag::{RelTag, SlruKind};
      31              : 
      32              : use postgres_ffi::dispatch_pgversion;
      33              : use postgres_ffi::pg_constants::{DEFAULTTABLESPACE_OID, GLOBALTABLESPACE_OID};
      34              : use postgres_ffi::pg_constants::{PGDATA_SPECIAL_FILES, PG_HBA};
      35              : use postgres_ffi::relfile_utils::{INIT_FORKNUM, MAIN_FORKNUM};
      36              : use postgres_ffi::XLogFileName;
      37              : use postgres_ffi::PG_TLI;
      38              : use postgres_ffi::{BLCKSZ, RELSEG_SIZE, WAL_SEGMENT_SIZE};
      39              : use utils::lsn::Lsn;
      40              : 
      41              : #[derive(Debug, thiserror::Error)]
      42              : pub enum BasebackupError {
      43              :     #[error("basebackup pageserver error {0:#}")]
      44              :     Server(#[from] anyhow::Error),
      45              :     #[error("basebackup client error {0:#} when {1}")]
      46              :     Client(#[source] io::Error, &'static str),
      47              : }
      48              : 
      49              : /// Create basebackup with non-rel data in it.
      50              : /// Only include relational data if 'full_backup' is true.
      51              : ///
      52              : /// Currently we use empty 'req_lsn' in two cases:
      53              : ///  * During the basebackup right after timeline creation
      54              : ///  * When working without safekeepers. In this situation it is important to match the lsn
      55              : ///    we are taking basebackup on with the lsn that is used in pageserver's walreceiver
      56              : ///    to start the replication.
      57            0 : pub async fn send_basebackup_tarball<'a, W>(
      58            0 :     write: &'a mut W,
      59            0 :     timeline: &'a Timeline,
      60            0 :     req_lsn: Option<Lsn>,
      61            0 :     prev_lsn: Option<Lsn>,
      62            0 :     full_backup: bool,
      63            0 :     replica: bool,
      64            0 :     ctx: &'a RequestContext,
      65            0 : ) -> Result<(), BasebackupError>
      66            0 : where
      67            0 :     W: AsyncWrite + Send + Sync + Unpin,
      68            0 : {
      69              :     // Compute postgres doesn't have any previous WAL files, but the first
      70              :     // record that it's going to write needs to include the LSN of the
      71              :     // previous record (xl_prev). We include prev_record_lsn in the
      72              :     // "zenith.signal" file, so that postgres can read it during startup.
      73              :     //
      74              :     // We don't keep full history of record boundaries in the page server,
      75              :     // however, only the predecessor of the latest record on each
      76              :     // timeline. So we can only provide prev_record_lsn when you take a
      77              :     // base backup at the end of the timeline, i.e. at last_record_lsn.
      78              :     // Even at the end of the timeline, we sometimes don't have a valid
      79              :     // prev_lsn value; that happens if the timeline was just branched from
      80              :     // an old LSN and it doesn't have any WAL of its own yet. We will set
      81              :     // prev_lsn to Lsn(0) if we cannot provide the correct value.
      82            0 :     let (backup_prev, backup_lsn) = if let Some(req_lsn) = req_lsn {
      83              :         // Backup was requested at a particular LSN. The caller should've
      84              :         // already checked that it's a valid LSN.
      85              : 
      86              :         // If the requested point is the end of the timeline, we can
      87              :         // provide prev_lsn. (get_last_record_rlsn() might return it as
      88              :         // zero, though, if no WAL has been generated on this timeline
      89              :         // yet.)
      90            0 :         let end_of_timeline = timeline.get_last_record_rlsn();
      91            0 :         if req_lsn == end_of_timeline.last {
      92            0 :             (end_of_timeline.prev, req_lsn)
      93              :         } else {
      94            0 :             (Lsn(0), req_lsn)
      95              :         }
      96              :     } else {
      97              :         // Backup was requested at end of the timeline.
      98            0 :         let end_of_timeline = timeline.get_last_record_rlsn();
      99            0 :         (end_of_timeline.prev, end_of_timeline.last)
     100              :     };
     101              : 
     102              :     // Consolidate the derived and the provided prev_lsn values
     103            0 :     let prev_lsn = if let Some(provided_prev_lsn) = prev_lsn {
     104            0 :         if backup_prev != Lsn(0) && backup_prev != provided_prev_lsn {
     105            0 :             return Err(BasebackupError::Server(anyhow!(
     106            0 :                 "backup_prev {backup_prev} != provided_prev_lsn {provided_prev_lsn}"
     107            0 :             )));
     108            0 :         }
     109            0 :         provided_prev_lsn
     110              :     } else {
     111            0 :         backup_prev
     112              :     };
     113              : 
     114            0 :     info!(
     115            0 :         "taking basebackup lsn={}, prev_lsn={} (full_backup={}, replica={})",
     116              :         backup_lsn, prev_lsn, full_backup, replica
     117              :     );
     118              : 
     119            0 :     let basebackup = Basebackup {
     120            0 :         ar: Builder::new_non_terminated(write),
     121            0 :         timeline,
     122            0 :         lsn: backup_lsn,
     123            0 :         prev_record_lsn: prev_lsn,
     124            0 :         full_backup,
     125            0 :         replica,
     126            0 :         ctx,
     127            0 :         io_concurrency: IoConcurrency::spawn_from_conf(
     128            0 :             timeline.conf,
     129            0 :             timeline
     130            0 :                 .gate
     131            0 :                 .enter()
     132            0 :                 .map_err(|e| BasebackupError::Server(e.into()))?,
     133              :         ),
     134              :     };
     135            0 :     basebackup
     136            0 :         .send_tarball()
     137            0 :         .instrument(info_span!("send_tarball", backup_lsn=%backup_lsn))
     138            0 :         .await
     139            0 : }
     140              : 
     141              : /// This is short-living object only for the time of tarball creation,
     142              : /// created mostly to avoid passing a lot of parameters between various functions
     143              : /// used for constructing tarball.
     144              : struct Basebackup<'a, W>
     145              : where
     146              :     W: AsyncWrite + Send + Sync + Unpin,
     147              : {
     148              :     ar: Builder<&'a mut W>,
     149              :     timeline: &'a Timeline,
     150              :     lsn: Lsn,
     151              :     prev_record_lsn: Lsn,
     152              :     full_backup: bool,
     153              :     replica: bool,
     154              :     ctx: &'a RequestContext,
     155              :     io_concurrency: IoConcurrency,
     156              : }
     157              : 
     158              : /// A sink that accepts SLRU blocks ordered by key and forwards
     159              : /// full segments to the archive.
     160              : struct SlruSegmentsBuilder<'a, 'b, W>
     161              : where
     162              :     W: AsyncWrite + Send + Sync + Unpin,
     163              : {
     164              :     ar: &'a mut Builder<&'b mut W>,
     165              :     buf: Vec<u8>,
     166              :     current_segment: Option<(SlruKind, u32)>,
     167              :     total_blocks: usize,
     168              : }
     169              : 
     170              : impl<'a, 'b, W> SlruSegmentsBuilder<'a, 'b, W>
     171              : where
     172              :     W: AsyncWrite + Send + Sync + Unpin,
     173              : {
     174            0 :     fn new(ar: &'a mut Builder<&'b mut W>) -> Self {
     175            0 :         Self {
     176            0 :             ar,
     177            0 :             buf: Vec::new(),
     178            0 :             current_segment: None,
     179            0 :             total_blocks: 0,
     180            0 :         }
     181            0 :     }
     182              : 
     183            0 :     async fn add_block(&mut self, key: &Key, block: Bytes) -> Result<(), BasebackupError> {
     184            0 :         let (kind, segno, _) = key.to_slru_block()?;
     185              : 
     186            0 :         match kind {
     187              :             SlruKind::Clog => {
     188            0 :                 if !(block.len() == BLCKSZ as usize || block.len() == BLCKSZ as usize + 8) {
     189            0 :                     return Err(BasebackupError::Server(anyhow!(
     190            0 :                         "invalid SlruKind::Clog record: block.len()={}",
     191            0 :                         block.len()
     192            0 :                     )));
     193            0 :                 }
     194              :             }
     195              :             SlruKind::MultiXactMembers | SlruKind::MultiXactOffsets => {
     196            0 :                 if block.len() != BLCKSZ as usize {
     197            0 :                     return Err(BasebackupError::Server(anyhow!(
     198            0 :                         "invalid {:?} record: block.len()={}",
     199            0 :                         kind,
     200            0 :                         block.len()
     201            0 :                     )));
     202            0 :                 }
     203              :             }
     204              :         }
     205              : 
     206            0 :         let segment = (kind, segno);
     207            0 :         match self.current_segment {
     208            0 :             None => {
     209            0 :                 self.current_segment = Some(segment);
     210            0 :                 self.buf
     211            0 :                     .extend_from_slice(block.slice(..BLCKSZ as usize).as_ref());
     212            0 :             }
     213            0 :             Some(current_seg) if current_seg == segment => {
     214            0 :                 self.buf
     215            0 :                     .extend_from_slice(block.slice(..BLCKSZ as usize).as_ref());
     216            0 :             }
     217              :             Some(_) => {
     218            0 :                 self.flush().await?;
     219              : 
     220            0 :                 self.current_segment = Some(segment);
     221            0 :                 self.buf
     222            0 :                     .extend_from_slice(block.slice(..BLCKSZ as usize).as_ref());
     223              :             }
     224              :         }
     225              : 
     226            0 :         Ok(())
     227            0 :     }
     228              : 
     229            0 :     async fn flush(&mut self) -> Result<(), BasebackupError> {
     230            0 :         let nblocks = self.buf.len() / BLCKSZ as usize;
     231            0 :         let (kind, segno) = self.current_segment.take().unwrap();
     232            0 :         let segname = format!("{}/{:>04X}", kind.to_str(), segno);
     233            0 :         let header = new_tar_header(&segname, self.buf.len() as u64)?;
     234            0 :         self.ar
     235            0 :             .append(&header, self.buf.as_slice())
     236            0 :             .await
     237            0 :             .map_err(|e| BasebackupError::Client(e, "flush"))?;
     238              : 
     239            0 :         self.total_blocks += nblocks;
     240            0 :         debug!("Added to basebackup slru {} relsize {}", segname, nblocks);
     241              : 
     242            0 :         self.buf.clear();
     243            0 : 
     244            0 :         Ok(())
     245            0 :     }
     246              : 
     247            0 :     async fn finish(mut self) -> Result<(), BasebackupError> {
     248            0 :         let res = if self.current_segment.is_none() || self.buf.is_empty() {
     249            0 :             Ok(())
     250              :         } else {
     251            0 :             self.flush().await
     252              :         };
     253              : 
     254            0 :         info!("Collected {} SLRU blocks", self.total_blocks);
     255              : 
     256            0 :         res
     257            0 :     }
     258              : }
     259              : 
     260              : impl<W> Basebackup<'_, W>
     261              : where
     262              :     W: AsyncWrite + Send + Sync + Unpin,
     263              : {
     264            0 :     async fn send_tarball(mut self) -> Result<(), BasebackupError> {
     265              :         // TODO include checksum
     266              : 
     267              :         // Construct the pg_control file from the persisted checkpoint and pg_control
     268              :         // information. But we only add this to the tarball at the end, so that if the
     269              :         // writing is interrupted half-way through, the resulting incomplete tarball will
     270              :         // be missing the pg_control file, which prevents PostgreSQL from starting up on
     271              :         // it. With proper error handling, you should never try to start up from an
     272              :         // incomplete basebackup in the first place, of course, but this is a nice little
     273              :         // extra safety measure.
     274            0 :         let checkpoint_bytes = self
     275            0 :             .timeline
     276            0 :             .get_checkpoint(self.lsn, self.ctx)
     277            0 :             .await
     278            0 :             .context("failed to get checkpoint bytes")?;
     279            0 :         let pg_control_bytes = self
     280            0 :             .timeline
     281            0 :             .get_control_file(self.lsn, self.ctx)
     282            0 :             .await
     283            0 :             .context("failed to get control bytes")?;
     284            0 :         let (pg_control_bytes, system_identifier, was_shutdown) =
     285            0 :             postgres_ffi::generate_pg_control(
     286            0 :                 &pg_control_bytes,
     287            0 :                 &checkpoint_bytes,
     288            0 :                 self.lsn,
     289            0 :                 self.timeline.pg_version,
     290            0 :             )?;
     291              : 
     292            0 :         let lazy_slru_download = self.timeline.get_lazy_slru_download() && !self.full_backup;
     293              : 
     294            0 :         let pgversion = self.timeline.pg_version;
     295            0 :         let subdirs = dispatch_pgversion!(pgversion, &pgv::bindings::PGDATA_SUBDIRS[..]);
     296              : 
     297              :         // Create pgdata subdirs structure
     298            0 :         for dir in subdirs.iter() {
     299            0 :             let header = new_tar_header_dir(dir)?;
     300            0 :             self.ar
     301            0 :                 .append(&header, io::empty())
     302            0 :                 .await
     303            0 :                 .map_err(|e| BasebackupError::Client(e, "send_tarball"))?;
     304              :         }
     305              : 
     306              :         // Send config files.
     307            0 :         for filepath in PGDATA_SPECIAL_FILES.iter() {
     308            0 :             if *filepath == "pg_hba.conf" {
     309            0 :                 let data = PG_HBA.as_bytes();
     310            0 :                 let header = new_tar_header(filepath, data.len() as u64)?;
     311            0 :                 self.ar
     312            0 :                     .append(&header, data)
     313            0 :                     .await
     314            0 :                     .map_err(|e| BasebackupError::Client(e, "send_tarball,pg_hba.conf"))?;
     315              :             } else {
     316            0 :                 let header = new_tar_header(filepath, 0)?;
     317            0 :                 self.ar
     318            0 :                     .append(&header, io::empty())
     319            0 :                     .await
     320            0 :                     .map_err(|e| BasebackupError::Client(e, "send_tarball,add_config_file"))?;
     321              :             }
     322              :         }
     323            0 :         if !lazy_slru_download {
     324              :             // Gather non-relational files from object storage pages.
     325            0 :             let slru_partitions = self
     326            0 :                 .timeline
     327            0 :                 .get_slru_keyspace(Version::Lsn(self.lsn), self.ctx)
     328            0 :                 .await
     329            0 :                 .map_err(|e| BasebackupError::Server(e.into()))?
     330            0 :                 .partition(
     331            0 :                     self.timeline.get_shard_identity(),
     332            0 :                     Timeline::MAX_GET_VECTORED_KEYS * BLCKSZ as u64,
     333            0 :                 );
     334            0 : 
     335            0 :             let mut slru_builder = SlruSegmentsBuilder::new(&mut self.ar);
     336              : 
     337            0 :             for part in slru_partitions.parts {
     338            0 :                 let blocks = self
     339            0 :                     .timeline
     340            0 :                     .get_vectored(part, self.lsn, self.io_concurrency.clone(), self.ctx)
     341            0 :                     .await
     342            0 :                     .map_err(|e| BasebackupError::Server(e.into()))?;
     343              : 
     344            0 :                 for (key, block) in blocks {
     345            0 :                     let block = block.map_err(|e| BasebackupError::Server(e.into()))?;
     346            0 :                     slru_builder.add_block(&key, block).await?;
     347              :                 }
     348              :             }
     349            0 :             slru_builder.finish().await?;
     350            0 :         }
     351              : 
     352            0 :         let mut min_restart_lsn: Lsn = Lsn::MAX;
     353              :         // Create tablespace directories
     354            0 :         for ((spcnode, dbnode), has_relmap_file) in self
     355            0 :             .timeline
     356            0 :             .list_dbdirs(self.lsn, self.ctx)
     357            0 :             .await
     358            0 :             .map_err(|e| BasebackupError::Server(e.into()))?
     359              :         {
     360            0 :             self.add_dbdir(spcnode, dbnode, has_relmap_file).await?;
     361              : 
     362              :             // If full backup is requested, include all relation files.
     363              :             // Otherwise only include init forks of unlogged relations.
     364            0 :             let rels = self
     365            0 :                 .timeline
     366            0 :                 .list_rels(spcnode, dbnode, Version::Lsn(self.lsn), self.ctx)
     367            0 :                 .await
     368            0 :                 .map_err(|e| BasebackupError::Server(e.into()))?;
     369            0 :             for &rel in rels.iter() {
     370              :                 // Send init fork as main fork to provide well formed empty
     371              :                 // contents of UNLOGGED relations. Postgres copies it in
     372              :                 // `reinit.c` during recovery.
     373            0 :                 if rel.forknum == INIT_FORKNUM {
     374              :                     // I doubt we need _init fork itself, but having it at least
     375              :                     // serves as a marker relation is unlogged.
     376            0 :                     self.add_rel(rel, rel).await?;
     377            0 :                     self.add_rel(rel, rel.with_forknum(MAIN_FORKNUM)).await?;
     378            0 :                     continue;
     379            0 :                 }
     380            0 : 
     381            0 :                 if self.full_backup {
     382            0 :                     if rel.forknum == MAIN_FORKNUM && rels.contains(&rel.with_forknum(INIT_FORKNUM))
     383              :                     {
     384              :                         // skip this, will include it when we reach the init fork
     385            0 :                         continue;
     386            0 :                     }
     387            0 :                     self.add_rel(rel, rel).await?;
     388            0 :                 }
     389              :             }
     390              :         }
     391              : 
     392            0 :         let start_time = Instant::now();
     393            0 :         let aux_files = self
     394            0 :             .timeline
     395            0 :             .list_aux_files(self.lsn, self.ctx, self.io_concurrency.clone())
     396            0 :             .await
     397            0 :             .map_err(|e| BasebackupError::Server(e.into()))?;
     398            0 :         let aux_scan_time = start_time.elapsed();
     399            0 :         let aux_estimated_size = aux_files
     400            0 :             .values()
     401            0 :             .map(|content| content.len())
     402            0 :             .sum::<usize>();
     403            0 :         info!(
     404            0 :             "Scanned {} aux files in {}ms, aux file content size = {}",
     405            0 :             aux_files.len(),
     406            0 :             aux_scan_time.as_millis(),
     407              :             aux_estimated_size
     408              :         );
     409              : 
     410            0 :         for (path, content) in aux_files {
     411            0 :             if path.starts_with("pg_replslot") {
     412              :                 // Do not create LR slots at standby because they are not used but prevent WAL truncation
     413            0 :                 if self.replica {
     414            0 :                     continue;
     415            0 :                 }
     416            0 :                 let offs = pg_constants::REPL_SLOT_ON_DISK_OFFSETOF_RESTART_LSN;
     417            0 :                 let restart_lsn = Lsn(u64::from_le_bytes(
     418            0 :                     content[offs..offs + 8].try_into().unwrap(),
     419            0 :                 ));
     420            0 :                 info!("Replication slot {} restart LSN={}", path, restart_lsn);
     421            0 :                 min_restart_lsn = Lsn::min(min_restart_lsn, restart_lsn);
     422            0 :             } else if path == "pg_logical/replorigin_checkpoint" {
     423              :                 // replorigin_checkoint is written only on compute shutdown, so it contains
     424              :                 // deteriorated values. So we generate our own version of this file for the particular LSN
     425              :                 // based on information about replorigins extracted from transaction commit records.
     426              :                 // In future we will not generate AUX record for "pg_logical/replorigin_checkpoint" at all,
     427              :                 // but now we should handle (skip) it for backward compatibility.
     428            0 :                 continue;
     429            0 :             } else if path == "pg_stat/pgstat.stat" && !was_shutdown {
     430              :                 // Drop statistic in case of abnormal termination, i.e. if we're not starting from the exact LSN
     431              :                 // of a shutdown checkpoint.
     432            0 :                 continue;
     433            0 :             }
     434            0 :             let header = new_tar_header(&path, content.len() as u64)?;
     435            0 :             self.ar
     436            0 :                 .append(&header, &*content)
     437            0 :                 .await
     438            0 :                 .map_err(|e| BasebackupError::Client(e, "send_tarball,add_aux_file"))?;
     439              :         }
     440              : 
     441            0 :         if min_restart_lsn != Lsn::MAX {
     442            0 :             info!(
     443            0 :                 "Min restart LSN for logical replication is {}",
     444              :                 min_restart_lsn
     445              :             );
     446            0 :             let data = min_restart_lsn.0.to_le_bytes();
     447            0 :             let header = new_tar_header("restart.lsn", data.len() as u64)?;
     448            0 :             self.ar
     449            0 :                 .append(&header, &data[..])
     450            0 :                 .await
     451            0 :                 .map_err(|e| BasebackupError::Client(e, "send_tarball,restart.lsn"))?;
     452            0 :         }
     453            0 :         for xid in self
     454            0 :             .timeline
     455            0 :             .list_twophase_files(self.lsn, self.ctx)
     456            0 :             .await
     457            0 :             .map_err(|e| BasebackupError::Server(e.into()))?
     458              :         {
     459            0 :             self.add_twophase_file(xid).await?;
     460              :         }
     461            0 :         let repl_origins = self
     462            0 :             .timeline
     463            0 :             .get_replorigins(self.lsn, self.ctx, self.io_concurrency.clone())
     464            0 :             .await
     465            0 :             .map_err(|e| BasebackupError::Server(e.into()))?;
     466            0 :         let n_origins = repl_origins.len();
     467            0 :         if n_origins != 0 {
     468              :             //
     469              :             // Construct "pg_logical/replorigin_checkpoint" file based on information about replication origins
     470              :             // extracted from transaction commit record. We are using this file to pass information about replication
     471              :             // origins to compute to allow logical replication to restart from proper point.
     472              :             //
     473            0 :             let mut content = Vec::with_capacity(n_origins * 16 + 8);
     474            0 :             content.extend_from_slice(&pg_constants::REPLICATION_STATE_MAGIC.to_le_bytes());
     475            0 :             for (origin_id, origin_lsn) in repl_origins {
     476            0 :                 content.extend_from_slice(&origin_id.to_le_bytes());
     477            0 :                 content.extend_from_slice(&[0u8; 6]); // align to 8 bytes
     478            0 :                 content.extend_from_slice(&origin_lsn.0.to_le_bytes());
     479            0 :             }
     480            0 :             let crc32 = crc32c::crc32c(&content);
     481            0 :             content.extend_from_slice(&crc32.to_le_bytes());
     482            0 :             let header = new_tar_header("pg_logical/replorigin_checkpoint", content.len() as u64)?;
     483            0 :             self.ar.append(&header, &*content).await.map_err(|e| {
     484            0 :                 BasebackupError::Client(e, "send_tarball,pg_logical/replorigin_checkpoint")
     485            0 :             })?;
     486            0 :         }
     487              : 
     488            0 :         fail_point!("basebackup-before-control-file", |_| {
     489            0 :             Err(BasebackupError::Server(anyhow!(
     490            0 :                 "failpoint basebackup-before-control-file"
     491            0 :             )))
     492            0 :         });
     493              : 
     494              :         // Last, add the pg_control file and bootstrap WAL segment.
     495            0 :         self.add_pgcontrol_file(pg_control_bytes, system_identifier)
     496            0 :             .await?;
     497            0 :         self.ar
     498            0 :             .finish()
     499            0 :             .await
     500            0 :             .map_err(|e| BasebackupError::Client(e, "send_tarball,finish"))?;
     501            0 :         debug!("all tarred up!");
     502            0 :         Ok(())
     503            0 :     }
     504              : 
     505              :     /// Add contents of relfilenode `src`, naming it as `dst`.
     506            0 :     async fn add_rel(&mut self, src: RelTag, dst: RelTag) -> Result<(), BasebackupError> {
     507            0 :         let nblocks = self
     508            0 :             .timeline
     509            0 :             .get_rel_size(src, Version::Lsn(self.lsn), self.ctx)
     510            0 :             .await
     511            0 :             .map_err(|e| BasebackupError::Server(e.into()))?;
     512              : 
     513              :         // If the relation is empty, create an empty file
     514            0 :         if nblocks == 0 {
     515            0 :             let file_name = dst.to_segfile_name(0);
     516            0 :             let header = new_tar_header(&file_name, 0)?;
     517            0 :             self.ar
     518            0 :                 .append(&header, io::empty())
     519            0 :                 .await
     520            0 :                 .map_err(|e| BasebackupError::Client(e, "add_rel,empty"))?;
     521            0 :             return Ok(());
     522            0 :         }
     523            0 : 
     524            0 :         // Add a file for each chunk of blocks (aka segment)
     525            0 :         let mut startblk = 0;
     526            0 :         let mut seg = 0;
     527            0 :         while startblk < nblocks {
     528            0 :             let endblk = std::cmp::min(startblk + RELSEG_SIZE, nblocks);
     529            0 : 
     530            0 :             let mut segment_data: Vec<u8> = vec![];
     531            0 :             for blknum in startblk..endblk {
     532            0 :                 let img = self
     533            0 :                     .timeline
     534            0 :                     // TODO: investigate using get_vectored for the entire startblk..endblk range.
     535            0 :                     // But this code path is not on the critical path for most basebackups (?).
     536            0 :                     .get(rel_block_to_key(src, blknum), self.lsn, self.ctx)
     537            0 :                     .await
     538            0 :                     .map_err(|e| BasebackupError::Server(e.into()))?;
     539            0 :                 segment_data.extend_from_slice(&img[..]);
     540              :             }
     541              : 
     542            0 :             let file_name = dst.to_segfile_name(seg as u32);
     543            0 :             let header = new_tar_header(&file_name, segment_data.len() as u64)?;
     544            0 :             self.ar
     545            0 :                 .append(&header, segment_data.as_slice())
     546            0 :                 .await
     547            0 :                 .map_err(|e| BasebackupError::Client(e, "add_rel,segment"))?;
     548              : 
     549            0 :             seg += 1;
     550            0 :             startblk = endblk;
     551              :         }
     552              : 
     553            0 :         Ok(())
     554            0 :     }
     555              : 
     556              :     //
     557              :     // Include database/tablespace directories.
     558              :     //
     559              :     // Each directory contains a PG_VERSION file, and the default database
     560              :     // directories also contain pg_filenode.map files.
     561              :     //
     562            0 :     async fn add_dbdir(
     563            0 :         &mut self,
     564            0 :         spcnode: u32,
     565            0 :         dbnode: u32,
     566            0 :         has_relmap_file: bool,
     567            0 :     ) -> Result<(), BasebackupError> {
     568            0 :         let relmap_img = if has_relmap_file {
     569            0 :             let img = self
     570            0 :                 .timeline
     571            0 :                 .get_relmap_file(spcnode, dbnode, Version::Lsn(self.lsn), self.ctx)
     572            0 :                 .await
     573            0 :                 .map_err(|e| BasebackupError::Server(e.into()))?;
     574              : 
     575            0 :             if img.len()
     576            0 :                 != dispatch_pgversion!(self.timeline.pg_version, pgv::bindings::SIZEOF_RELMAPFILE)
     577              :             {
     578            0 :                 return Err(BasebackupError::Server(anyhow!(
     579            0 :                     "img.len() != SIZE_OF_RELMAPFILE, img.len()={}",
     580            0 :                     img.len(),
     581            0 :                 )));
     582            0 :             }
     583            0 : 
     584            0 :             Some(img)
     585              :         } else {
     586            0 :             None
     587              :         };
     588              : 
     589            0 :         if spcnode == GLOBALTABLESPACE_OID {
     590            0 :             let pg_version_str = match self.timeline.pg_version {
     591            0 :                 14 | 15 => self.timeline.pg_version.to_string(),
     592            0 :                 ver => format!("{ver}\x0A"),
     593              :             };
     594            0 :             let header = new_tar_header("PG_VERSION", pg_version_str.len() as u64)?;
     595            0 :             self.ar
     596            0 :                 .append(&header, pg_version_str.as_bytes())
     597            0 :                 .await
     598            0 :                 .map_err(|e| BasebackupError::Client(e, "add_dbdir,PG_VERSION"))?;
     599              : 
     600            0 :             info!("timeline.pg_version {}", self.timeline.pg_version);
     601              : 
     602            0 :             if let Some(img) = relmap_img {
     603              :                 // filenode map for global tablespace
     604            0 :                 let header = new_tar_header("global/pg_filenode.map", img.len() as u64)?;
     605            0 :                 self.ar
     606            0 :                     .append(&header, &img[..])
     607            0 :                     .await
     608            0 :                     .map_err(|e| BasebackupError::Client(e, "add_dbdir,global/pg_filenode.map"))?;
     609              :             } else {
     610            0 :                 warn!("global/pg_filenode.map is missing");
     611              :             }
     612              :         } else {
     613              :             // User defined tablespaces are not supported. However, as
     614              :             // a special case, if a tablespace/db directory is
     615              :             // completely empty, we can leave it out altogether. This
     616              :             // makes taking a base backup after the 'tablespace'
     617              :             // regression test pass, because the test drops the
     618              :             // created tablespaces after the tests.
     619              :             //
     620              :             // FIXME: this wouldn't be necessary, if we handled
     621              :             // XLOG_TBLSPC_DROP records. But we probably should just
     622              :             // throw an error on CREATE TABLESPACE in the first place.
     623            0 :             if !has_relmap_file
     624            0 :                 && self
     625            0 :                     .timeline
     626            0 :                     .list_rels(spcnode, dbnode, Version::Lsn(self.lsn), self.ctx)
     627            0 :                     .await
     628            0 :                     .map_err(|e| BasebackupError::Server(e.into()))?
     629            0 :                     .is_empty()
     630              :             {
     631            0 :                 return Ok(());
     632            0 :             }
     633            0 :             // User defined tablespaces are not supported
     634            0 :             if spcnode != DEFAULTTABLESPACE_OID {
     635            0 :                 return Err(BasebackupError::Server(anyhow!(
     636            0 :                     "spcnode != DEFAULTTABLESPACE_OID, spcnode={spcnode}"
     637            0 :                 )));
     638            0 :             }
     639            0 : 
     640            0 :             // Append dir path for each database
     641            0 :             let path = format!("base/{}", dbnode);
     642            0 :             let header = new_tar_header_dir(&path)?;
     643            0 :             self.ar
     644            0 :                 .append(&header, io::empty())
     645            0 :                 .await
     646            0 :                 .map_err(|e| BasebackupError::Client(e, "add_dbdir,base"))?;
     647              : 
     648            0 :             if let Some(img) = relmap_img {
     649            0 :                 let dst_path = format!("base/{}/PG_VERSION", dbnode);
     650              : 
     651            0 :                 let pg_version_str = match self.timeline.pg_version {
     652            0 :                     14 | 15 => self.timeline.pg_version.to_string(),
     653            0 :                     ver => format!("{ver}\x0A"),
     654              :                 };
     655            0 :                 let header = new_tar_header(&dst_path, pg_version_str.len() as u64)?;
     656            0 :                 self.ar
     657            0 :                     .append(&header, pg_version_str.as_bytes())
     658            0 :                     .await
     659            0 :                     .map_err(|e| BasebackupError::Client(e, "add_dbdir,base/PG_VERSION"))?;
     660              : 
     661            0 :                 let relmap_path = format!("base/{}/pg_filenode.map", dbnode);
     662            0 :                 let header = new_tar_header(&relmap_path, img.len() as u64)?;
     663            0 :                 self.ar
     664            0 :                     .append(&header, &img[..])
     665            0 :                     .await
     666            0 :                     .map_err(|e| BasebackupError::Client(e, "add_dbdir,base/pg_filenode.map"))?;
     667            0 :             }
     668              :         };
     669            0 :         Ok(())
     670            0 :     }
     671              : 
     672              :     //
     673              :     // Extract twophase state files
     674              :     //
     675            0 :     async fn add_twophase_file(&mut self, xid: u64) -> Result<(), BasebackupError> {
     676            0 :         let img = self
     677            0 :             .timeline
     678            0 :             .get_twophase_file(xid, self.lsn, self.ctx)
     679            0 :             .await
     680            0 :             .map_err(|e| BasebackupError::Server(e.into()))?;
     681              : 
     682            0 :         let mut buf = BytesMut::new();
     683            0 :         buf.extend_from_slice(&img[..]);
     684            0 :         let crc = crc32c::crc32c(&img[..]);
     685            0 :         buf.put_u32_le(crc);
     686            0 :         let path = if self.timeline.pg_version < 17 {
     687            0 :             format!("pg_twophase/{:>08X}", xid)
     688              :         } else {
     689            0 :             format!("pg_twophase/{:>016X}", xid)
     690              :         };
     691            0 :         let header = new_tar_header(&path, buf.len() as u64)?;
     692            0 :         self.ar
     693            0 :             .append(&header, &buf[..])
     694            0 :             .await
     695            0 :             .map_err(|e| BasebackupError::Client(e, "add_twophase_file"))?;
     696              : 
     697            0 :         Ok(())
     698            0 :     }
     699              : 
     700              :     //
     701              :     // Add generated pg_control file and bootstrap WAL segment.
     702              :     // Also send zenith.signal file with extra bootstrap data.
     703              :     //
     704            0 :     async fn add_pgcontrol_file(
     705            0 :         &mut self,
     706            0 :         pg_control_bytes: Bytes,
     707            0 :         system_identifier: u64,
     708            0 :     ) -> Result<(), BasebackupError> {
     709            0 :         // add zenith.signal file
     710            0 :         let mut zenith_signal = String::new();
     711            0 :         if self.prev_record_lsn == Lsn(0) {
     712            0 :             if self.timeline.is_ancestor_lsn(self.lsn) {
     713            0 :                 write!(zenith_signal, "PREV LSN: none")
     714            0 :                     .map_err(|e| BasebackupError::Server(e.into()))?;
     715              :             } else {
     716            0 :                 write!(zenith_signal, "PREV LSN: invalid")
     717            0 :                     .map_err(|e| BasebackupError::Server(e.into()))?;
     718              :             }
     719              :         } else {
     720            0 :             write!(zenith_signal, "PREV LSN: {}", self.prev_record_lsn)
     721            0 :                 .map_err(|e| BasebackupError::Server(e.into()))?;
     722              :         }
     723            0 :         self.ar
     724            0 :             .append(
     725            0 :                 &new_tar_header("zenith.signal", zenith_signal.len() as u64)?,
     726            0 :                 zenith_signal.as_bytes(),
     727            0 :             )
     728            0 :             .await
     729            0 :             .map_err(|e| BasebackupError::Client(e, "add_pgcontrol_file,zenith.signal"))?;
     730              : 
     731              :         //send pg_control
     732            0 :         let header = new_tar_header("global/pg_control", pg_control_bytes.len() as u64)?;
     733            0 :         self.ar
     734            0 :             .append(&header, &pg_control_bytes[..])
     735            0 :             .await
     736            0 :             .map_err(|e| BasebackupError::Client(e, "add_pgcontrol_file,pg_control"))?;
     737              : 
     738              :         //send wal segment
     739            0 :         let segno = self.lsn.segment_number(WAL_SEGMENT_SIZE);
     740            0 :         let wal_file_name = XLogFileName(PG_TLI, segno, WAL_SEGMENT_SIZE);
     741            0 :         let wal_file_path = format!("pg_wal/{}", wal_file_name);
     742            0 :         let header = new_tar_header(&wal_file_path, WAL_SEGMENT_SIZE as u64)?;
     743              : 
     744            0 :         let wal_seg = postgres_ffi::generate_wal_segment(
     745            0 :             segno,
     746            0 :             system_identifier,
     747            0 :             self.timeline.pg_version,
     748            0 :             self.lsn,
     749            0 :         )
     750            0 :         .map_err(|e| anyhow!(e).context("Failed generating wal segment"))?;
     751            0 :         if wal_seg.len() != WAL_SEGMENT_SIZE {
     752            0 :             return Err(BasebackupError::Server(anyhow!(
     753            0 :                 "wal_seg.len() != WAL_SEGMENT_SIZE, wal_seg.len()={}",
     754            0 :                 wal_seg.len()
     755            0 :             )));
     756            0 :         }
     757            0 :         self.ar
     758            0 :             .append(&header, &wal_seg[..])
     759            0 :             .await
     760            0 :             .map_err(|e| BasebackupError::Client(e, "add_pgcontrol_file,wal_segment"))?;
     761            0 :         Ok(())
     762            0 :     }
     763              : }
     764              : 
     765              : //
     766              : // Create new tarball entry header
     767              : //
     768            0 : fn new_tar_header(path: &str, size: u64) -> anyhow::Result<Header> {
     769            0 :     let mut header = Header::new_gnu();
     770            0 :     header.set_size(size);
     771            0 :     header.set_path(path)?;
     772            0 :     header.set_mode(0b110000000); // -rw-------
     773            0 :     header.set_mtime(
     774            0 :         // use currenttime as last modified time
     775            0 :         SystemTime::now()
     776            0 :             .duration_since(SystemTime::UNIX_EPOCH)
     777            0 :             .unwrap()
     778            0 :             .as_secs(),
     779            0 :     );
     780            0 :     header.set_cksum();
     781            0 :     Ok(header)
     782            0 : }
     783              : 
     784            0 : fn new_tar_header_dir(path: &str) -> anyhow::Result<Header> {
     785            0 :     let mut header = Header::new_gnu();
     786            0 :     header.set_size(0);
     787            0 :     header.set_path(path)?;
     788            0 :     header.set_mode(0o755); // -rw-------
     789            0 :     header.set_entry_type(EntryType::dir());
     790            0 :     header.set_mtime(
     791            0 :         // use currenttime as last modified time
     792            0 :         SystemTime::now()
     793            0 :             .duration_since(SystemTime::UNIX_EPOCH)
     794            0 :             .unwrap()
     795            0 :             .as_secs(),
     796            0 :     );
     797            0 :     header.set_cksum();
     798            0 :     Ok(header)
     799            0 : }
        

Generated by: LCOV version 2.1-beta