LCOV - code coverage report
Current view: top level - pageserver/src - basebackup.rs (source / functions) Coverage Total Hit
Test: 1d18b743246dcf78c27c0bad0234a4c0da6fde89.info Lines: 0.0 % 506 0
Test Date: 2025-02-14 00:11:37 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::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            0 :         let lazy_slru_download = self.timeline.get_lazy_slru_download() && !self.full_backup;
     268              : 
     269            0 :         let pgversion = self.timeline.pg_version;
     270            0 :         let subdirs = dispatch_pgversion!(pgversion, &pgv::bindings::PGDATA_SUBDIRS[..]);
     271              : 
     272              :         // Create pgdata subdirs structure
     273            0 :         for dir in subdirs.iter() {
     274            0 :             let header = new_tar_header_dir(dir)?;
     275            0 :             self.ar
     276            0 :                 .append(&header, io::empty())
     277            0 :                 .await
     278            0 :                 .map_err(|e| BasebackupError::Client(e, "send_tarball"))?;
     279              :         }
     280              : 
     281              :         // Send config files.
     282            0 :         for filepath in PGDATA_SPECIAL_FILES.iter() {
     283            0 :             if *filepath == "pg_hba.conf" {
     284            0 :                 let data = PG_HBA.as_bytes();
     285            0 :                 let header = new_tar_header(filepath, data.len() as u64)?;
     286            0 :                 self.ar
     287            0 :                     .append(&header, data)
     288            0 :                     .await
     289            0 :                     .map_err(|e| BasebackupError::Client(e, "send_tarball,pg_hba.conf"))?;
     290              :             } else {
     291            0 :                 let header = new_tar_header(filepath, 0)?;
     292            0 :                 self.ar
     293            0 :                     .append(&header, io::empty())
     294            0 :                     .await
     295            0 :                     .map_err(|e| BasebackupError::Client(e, "send_tarball,add_config_file"))?;
     296              :             }
     297              :         }
     298            0 :         if !lazy_slru_download {
     299              :             // Gather non-relational files from object storage pages.
     300            0 :             let slru_partitions = self
     301            0 :                 .timeline
     302            0 :                 .get_slru_keyspace(Version::Lsn(self.lsn), self.ctx)
     303            0 :                 .await
     304            0 :                 .map_err(|e| BasebackupError::Server(e.into()))?
     305            0 :                 .partition(
     306            0 :                     self.timeline.get_shard_identity(),
     307            0 :                     Timeline::MAX_GET_VECTORED_KEYS * BLCKSZ as u64,
     308            0 :                 );
     309            0 : 
     310            0 :             let mut slru_builder = SlruSegmentsBuilder::new(&mut self.ar);
     311              : 
     312            0 :             for part in slru_partitions.parts {
     313            0 :                 let blocks = self
     314            0 :                     .timeline
     315            0 :                     .get_vectored(part, self.lsn, self.io_concurrency.clone(), self.ctx)
     316            0 :                     .await
     317            0 :                     .map_err(|e| BasebackupError::Server(e.into()))?;
     318              : 
     319            0 :                 for (key, block) in blocks {
     320            0 :                     let block = block.map_err(|e| BasebackupError::Server(e.into()))?;
     321            0 :                     slru_builder.add_block(&key, block).await?;
     322              :                 }
     323              :             }
     324            0 :             slru_builder.finish().await?;
     325            0 :         }
     326              : 
     327            0 :         let mut min_restart_lsn: Lsn = Lsn::MAX;
     328              :         // Create tablespace directories
     329            0 :         for ((spcnode, dbnode), has_relmap_file) in self
     330            0 :             .timeline
     331            0 :             .list_dbdirs(self.lsn, self.ctx)
     332            0 :             .await
     333            0 :             .map_err(|e| BasebackupError::Server(e.into()))?
     334              :         {
     335            0 :             self.add_dbdir(spcnode, dbnode, has_relmap_file).await?;
     336              : 
     337              :             // If full backup is requested, include all relation files.
     338              :             // Otherwise only include init forks of unlogged relations.
     339            0 :             let rels = self
     340            0 :                 .timeline
     341            0 :                 .list_rels(spcnode, dbnode, Version::Lsn(self.lsn), self.ctx)
     342            0 :                 .await
     343            0 :                 .map_err(|e| BasebackupError::Server(e.into()))?;
     344            0 :             for &rel in rels.iter() {
     345              :                 // Send init fork as main fork to provide well formed empty
     346              :                 // contents of UNLOGGED relations. Postgres copies it in
     347              :                 // `reinit.c` during recovery.
     348            0 :                 if rel.forknum == INIT_FORKNUM {
     349              :                     // I doubt we need _init fork itself, but having it at least
     350              :                     // serves as a marker relation is unlogged.
     351            0 :                     self.add_rel(rel, rel).await?;
     352            0 :                     self.add_rel(rel, rel.with_forknum(MAIN_FORKNUM)).await?;
     353            0 :                     continue;
     354            0 :                 }
     355            0 : 
     356            0 :                 if self.full_backup {
     357            0 :                     if rel.forknum == MAIN_FORKNUM && rels.contains(&rel.with_forknum(INIT_FORKNUM))
     358              :                     {
     359              :                         // skip this, will include it when we reach the init fork
     360            0 :                         continue;
     361            0 :                     }
     362            0 :                     self.add_rel(rel, rel).await?;
     363            0 :                 }
     364              :             }
     365              :         }
     366              : 
     367            0 :         let start_time = Instant::now();
     368            0 :         let aux_files = self
     369            0 :             .timeline
     370            0 :             .list_aux_files(self.lsn, self.ctx, self.io_concurrency.clone())
     371            0 :             .await
     372            0 :             .map_err(|e| BasebackupError::Server(e.into()))?;
     373            0 :         let aux_scan_time = start_time.elapsed();
     374            0 :         let aux_estimated_size = aux_files
     375            0 :             .values()
     376            0 :             .map(|content| content.len())
     377            0 :             .sum::<usize>();
     378            0 :         info!(
     379            0 :             "Scanned {} aux files in {}ms, aux file content size = {}",
     380            0 :             aux_files.len(),
     381            0 :             aux_scan_time.as_millis(),
     382              :             aux_estimated_size
     383              :         );
     384              : 
     385            0 :         for (path, content) in aux_files {
     386            0 :             if path.starts_with("pg_replslot") {
     387              :                 // Do not create LR slots at standby because they are not used but prevent WAL truncation
     388            0 :                 if self.replica {
     389            0 :                     continue;
     390            0 :                 }
     391            0 :                 let offs = pg_constants::REPL_SLOT_ON_DISK_OFFSETOF_RESTART_LSN;
     392            0 :                 let restart_lsn = Lsn(u64::from_le_bytes(
     393            0 :                     content[offs..offs + 8].try_into().unwrap(),
     394            0 :                 ));
     395            0 :                 info!("Replication slot {} restart LSN={}", path, restart_lsn);
     396            0 :                 min_restart_lsn = Lsn::min(min_restart_lsn, restart_lsn);
     397            0 :             } else if path == "pg_logical/replorigin_checkpoint" {
     398              :                 // replorigin_checkoint is written only on compute shutdown, so it contains
     399              :                 // deteriorated values. So we generate our own version of this file for the particular LSN
     400              :                 // based on information about replorigins extracted from transaction commit records.
     401              :                 // In future we will not generate AUX record for "pg_logical/replorigin_checkpoint" at all,
     402              :                 // but now we should handle (skip) it for backward compatibility.
     403            0 :                 continue;
     404            0 :             }
     405            0 :             let header = new_tar_header(&path, content.len() as u64)?;
     406            0 :             self.ar
     407            0 :                 .append(&header, &*content)
     408            0 :                 .await
     409            0 :                 .map_err(|e| BasebackupError::Client(e, "send_tarball,add_aux_file"))?;
     410              :         }
     411              : 
     412            0 :         if min_restart_lsn != Lsn::MAX {
     413            0 :             info!(
     414            0 :                 "Min restart LSN for logical replication is {}",
     415              :                 min_restart_lsn
     416              :             );
     417            0 :             let data = min_restart_lsn.0.to_le_bytes();
     418            0 :             let header = new_tar_header("restart.lsn", data.len() as u64)?;
     419            0 :             self.ar
     420            0 :                 .append(&header, &data[..])
     421            0 :                 .await
     422            0 :                 .map_err(|e| BasebackupError::Client(e, "send_tarball,restart.lsn"))?;
     423            0 :         }
     424            0 :         for xid in self
     425            0 :             .timeline
     426            0 :             .list_twophase_files(self.lsn, self.ctx)
     427            0 :             .await
     428            0 :             .map_err(|e| BasebackupError::Server(e.into()))?
     429              :         {
     430            0 :             self.add_twophase_file(xid).await?;
     431              :         }
     432            0 :         let repl_origins = self
     433            0 :             .timeline
     434            0 :             .get_replorigins(self.lsn, self.ctx, self.io_concurrency.clone())
     435            0 :             .await
     436            0 :             .map_err(|e| BasebackupError::Server(e.into()))?;
     437            0 :         let n_origins = repl_origins.len();
     438            0 :         if n_origins != 0 {
     439              :             //
     440              :             // Construct "pg_logical/replorigin_checkpoint" file based on information about replication origins
     441              :             // extracted from transaction commit record. We are using this file to pass information about replication
     442              :             // origins to compute to allow logical replication to restart from proper point.
     443              :             //
     444            0 :             let mut content = Vec::with_capacity(n_origins * 16 + 8);
     445            0 :             content.extend_from_slice(&pg_constants::REPLICATION_STATE_MAGIC.to_le_bytes());
     446            0 :             for (origin_id, origin_lsn) in repl_origins {
     447            0 :                 content.extend_from_slice(&origin_id.to_le_bytes());
     448            0 :                 content.extend_from_slice(&[0u8; 6]); // align to 8 bytes
     449            0 :                 content.extend_from_slice(&origin_lsn.0.to_le_bytes());
     450            0 :             }
     451            0 :             let crc32 = crc32c::crc32c(&content);
     452            0 :             content.extend_from_slice(&crc32.to_le_bytes());
     453            0 :             let header = new_tar_header("pg_logical/replorigin_checkpoint", content.len() as u64)?;
     454            0 :             self.ar.append(&header, &*content).await.map_err(|e| {
     455            0 :                 BasebackupError::Client(e, "send_tarball,pg_logical/replorigin_checkpoint")
     456            0 :             })?;
     457            0 :         }
     458              : 
     459            0 :         fail_point!("basebackup-before-control-file", |_| {
     460            0 :             Err(BasebackupError::Server(anyhow!(
     461            0 :                 "failpoint basebackup-before-control-file"
     462            0 :             )))
     463            0 :         });
     464              : 
     465              :         // Generate pg_control and bootstrap WAL segment.
     466            0 :         self.add_pgcontrol_file().await?;
     467            0 :         self.ar
     468            0 :             .finish()
     469            0 :             .await
     470            0 :             .map_err(|e| BasebackupError::Client(e, "send_tarball,finish"))?;
     471            0 :         debug!("all tarred up!");
     472            0 :         Ok(())
     473            0 :     }
     474              : 
     475              :     /// Add contents of relfilenode `src`, naming it as `dst`.
     476            0 :     async fn add_rel(&mut self, src: RelTag, dst: RelTag) -> Result<(), BasebackupError> {
     477            0 :         let nblocks = self
     478            0 :             .timeline
     479            0 :             .get_rel_size(src, Version::Lsn(self.lsn), self.ctx)
     480            0 :             .await
     481            0 :             .map_err(|e| BasebackupError::Server(e.into()))?;
     482              : 
     483              :         // If the relation is empty, create an empty file
     484            0 :         if nblocks == 0 {
     485            0 :             let file_name = dst.to_segfile_name(0);
     486            0 :             let header = new_tar_header(&file_name, 0)?;
     487            0 :             self.ar
     488            0 :                 .append(&header, io::empty())
     489            0 :                 .await
     490            0 :                 .map_err(|e| BasebackupError::Client(e, "add_rel,empty"))?;
     491            0 :             return Ok(());
     492            0 :         }
     493            0 : 
     494            0 :         // Add a file for each chunk of blocks (aka segment)
     495            0 :         let mut startblk = 0;
     496            0 :         let mut seg = 0;
     497            0 :         while startblk < nblocks {
     498            0 :             let endblk = std::cmp::min(startblk + RELSEG_SIZE, nblocks);
     499            0 : 
     500            0 :             let mut segment_data: Vec<u8> = vec![];
     501            0 :             for blknum in startblk..endblk {
     502            0 :                 let img = self
     503            0 :                     .timeline
     504            0 :                     .get_rel_page_at_lsn(
     505            0 :                         src,
     506            0 :                         blknum,
     507            0 :                         Version::Lsn(self.lsn),
     508            0 :                         self.ctx,
     509            0 :                         self.io_concurrency.clone(),
     510            0 :                     )
     511            0 :                     .await
     512            0 :                     .map_err(|e| BasebackupError::Server(e.into()))?;
     513            0 :                 segment_data.extend_from_slice(&img[..]);
     514              :             }
     515              : 
     516            0 :             let file_name = dst.to_segfile_name(seg as u32);
     517            0 :             let header = new_tar_header(&file_name, segment_data.len() as u64)?;
     518            0 :             self.ar
     519            0 :                 .append(&header, segment_data.as_slice())
     520            0 :                 .await
     521            0 :                 .map_err(|e| BasebackupError::Client(e, "add_rel,segment"))?;
     522              : 
     523            0 :             seg += 1;
     524            0 :             startblk = endblk;
     525              :         }
     526              : 
     527            0 :         Ok(())
     528            0 :     }
     529              : 
     530              :     //
     531              :     // Include database/tablespace directories.
     532              :     //
     533              :     // Each directory contains a PG_VERSION file, and the default database
     534              :     // directories also contain pg_filenode.map files.
     535              :     //
     536            0 :     async fn add_dbdir(
     537            0 :         &mut self,
     538            0 :         spcnode: u32,
     539            0 :         dbnode: u32,
     540            0 :         has_relmap_file: bool,
     541            0 :     ) -> Result<(), BasebackupError> {
     542            0 :         let relmap_img = if has_relmap_file {
     543            0 :             let img = self
     544            0 :                 .timeline
     545            0 :                 .get_relmap_file(spcnode, dbnode, Version::Lsn(self.lsn), self.ctx)
     546            0 :                 .await
     547            0 :                 .map_err(|e| BasebackupError::Server(e.into()))?;
     548              : 
     549            0 :             if img.len()
     550            0 :                 != dispatch_pgversion!(self.timeline.pg_version, pgv::bindings::SIZEOF_RELMAPFILE)
     551              :             {
     552            0 :                 return Err(BasebackupError::Server(anyhow!(
     553            0 :                     "img.len() != SIZE_OF_RELMAPFILE, img.len()={}",
     554            0 :                     img.len(),
     555            0 :                 )));
     556            0 :             }
     557            0 : 
     558            0 :             Some(img)
     559              :         } else {
     560            0 :             None
     561              :         };
     562              : 
     563            0 :         if spcnode == GLOBALTABLESPACE_OID {
     564            0 :             let pg_version_str = match self.timeline.pg_version {
     565            0 :                 14 | 15 => self.timeline.pg_version.to_string(),
     566            0 :                 ver => format!("{ver}\x0A"),
     567              :             };
     568            0 :             let header = new_tar_header("PG_VERSION", pg_version_str.len() as u64)?;
     569            0 :             self.ar
     570            0 :                 .append(&header, pg_version_str.as_bytes())
     571            0 :                 .await
     572            0 :                 .map_err(|e| BasebackupError::Client(e, "add_dbdir,PG_VERSION"))?;
     573              : 
     574            0 :             info!("timeline.pg_version {}", self.timeline.pg_version);
     575              : 
     576            0 :             if let Some(img) = relmap_img {
     577              :                 // filenode map for global tablespace
     578            0 :                 let header = new_tar_header("global/pg_filenode.map", img.len() as u64)?;
     579            0 :                 self.ar
     580            0 :                     .append(&header, &img[..])
     581            0 :                     .await
     582            0 :                     .map_err(|e| BasebackupError::Client(e, "add_dbdir,global/pg_filenode.map"))?;
     583              :             } else {
     584            0 :                 warn!("global/pg_filenode.map is missing");
     585              :             }
     586              :         } else {
     587              :             // User defined tablespaces are not supported. However, as
     588              :             // a special case, if a tablespace/db directory is
     589              :             // completely empty, we can leave it out altogether. This
     590              :             // makes taking a base backup after the 'tablespace'
     591              :             // regression test pass, because the test drops the
     592              :             // created tablespaces after the tests.
     593              :             //
     594              :             // FIXME: this wouldn't be necessary, if we handled
     595              :             // XLOG_TBLSPC_DROP records. But we probably should just
     596              :             // throw an error on CREATE TABLESPACE in the first place.
     597            0 :             if !has_relmap_file
     598            0 :                 && self
     599            0 :                     .timeline
     600            0 :                     .list_rels(spcnode, dbnode, Version::Lsn(self.lsn), self.ctx)
     601            0 :                     .await
     602            0 :                     .map_err(|e| BasebackupError::Server(e.into()))?
     603            0 :                     .is_empty()
     604              :             {
     605            0 :                 return Ok(());
     606            0 :             }
     607            0 :             // User defined tablespaces are not supported
     608            0 :             if spcnode != DEFAULTTABLESPACE_OID {
     609            0 :                 return Err(BasebackupError::Server(anyhow!(
     610            0 :                     "spcnode != DEFAULTTABLESPACE_OID, spcnode={spcnode}"
     611            0 :                 )));
     612            0 :             }
     613            0 : 
     614            0 :             // Append dir path for each database
     615            0 :             let path = format!("base/{}", dbnode);
     616            0 :             let header = new_tar_header_dir(&path)?;
     617            0 :             self.ar
     618            0 :                 .append(&header, io::empty())
     619            0 :                 .await
     620            0 :                 .map_err(|e| BasebackupError::Client(e, "add_dbdir,base"))?;
     621              : 
     622            0 :             if let Some(img) = relmap_img {
     623            0 :                 let dst_path = format!("base/{}/PG_VERSION", dbnode);
     624              : 
     625            0 :                 let pg_version_str = match self.timeline.pg_version {
     626            0 :                     14 | 15 => self.timeline.pg_version.to_string(),
     627            0 :                     ver => format!("{ver}\x0A"),
     628              :                 };
     629            0 :                 let header = new_tar_header(&dst_path, pg_version_str.len() as u64)?;
     630            0 :                 self.ar
     631            0 :                     .append(&header, pg_version_str.as_bytes())
     632            0 :                     .await
     633            0 :                     .map_err(|e| BasebackupError::Client(e, "add_dbdir,base/PG_VERSION"))?;
     634              : 
     635            0 :                 let relmap_path = format!("base/{}/pg_filenode.map", dbnode);
     636            0 :                 let header = new_tar_header(&relmap_path, img.len() as u64)?;
     637            0 :                 self.ar
     638            0 :                     .append(&header, &img[..])
     639            0 :                     .await
     640            0 :                     .map_err(|e| BasebackupError::Client(e, "add_dbdir,base/pg_filenode.map"))?;
     641            0 :             }
     642              :         };
     643            0 :         Ok(())
     644            0 :     }
     645              : 
     646              :     //
     647              :     // Extract twophase state files
     648              :     //
     649            0 :     async fn add_twophase_file(&mut self, xid: u64) -> Result<(), BasebackupError> {
     650            0 :         let img = self
     651            0 :             .timeline
     652            0 :             .get_twophase_file(xid, self.lsn, self.ctx)
     653            0 :             .await
     654            0 :             .map_err(|e| BasebackupError::Server(e.into()))?;
     655              : 
     656            0 :         let mut buf = BytesMut::new();
     657            0 :         buf.extend_from_slice(&img[..]);
     658            0 :         let crc = crc32c::crc32c(&img[..]);
     659            0 :         buf.put_u32_le(crc);
     660            0 :         let path = if self.timeline.pg_version < 17 {
     661            0 :             format!("pg_twophase/{:>08X}", xid)
     662              :         } else {
     663            0 :             format!("pg_twophase/{:>016X}", xid)
     664              :         };
     665            0 :         let header = new_tar_header(&path, buf.len() as u64)?;
     666            0 :         self.ar
     667            0 :             .append(&header, &buf[..])
     668            0 :             .await
     669            0 :             .map_err(|e| BasebackupError::Client(e, "add_twophase_file"))?;
     670              : 
     671            0 :         Ok(())
     672            0 :     }
     673              : 
     674              :     //
     675              :     // Add generated pg_control file and bootstrap WAL segment.
     676              :     // Also send zenith.signal file with extra bootstrap data.
     677              :     //
     678            0 :     async fn add_pgcontrol_file(&mut self) -> Result<(), BasebackupError> {
     679            0 :         // add zenith.signal file
     680            0 :         let mut zenith_signal = String::new();
     681            0 :         if self.prev_record_lsn == Lsn(0) {
     682            0 :             if self.timeline.is_ancestor_lsn(self.lsn) {
     683            0 :                 write!(zenith_signal, "PREV LSN: none")
     684            0 :                     .map_err(|e| BasebackupError::Server(e.into()))?;
     685              :             } else {
     686            0 :                 write!(zenith_signal, "PREV LSN: invalid")
     687            0 :                     .map_err(|e| BasebackupError::Server(e.into()))?;
     688              :             }
     689              :         } else {
     690            0 :             write!(zenith_signal, "PREV LSN: {}", self.prev_record_lsn)
     691            0 :                 .map_err(|e| BasebackupError::Server(e.into()))?;
     692              :         }
     693            0 :         self.ar
     694            0 :             .append(
     695            0 :                 &new_tar_header("zenith.signal", zenith_signal.len() as u64)?,
     696            0 :                 zenith_signal.as_bytes(),
     697            0 :             )
     698            0 :             .await
     699            0 :             .map_err(|e| BasebackupError::Client(e, "add_pgcontrol_file,zenith.signal"))?;
     700              : 
     701            0 :         let checkpoint_bytes = self
     702            0 :             .timeline
     703            0 :             .get_checkpoint(self.lsn, self.ctx)
     704            0 :             .await
     705            0 :             .context("failed to get checkpoint bytes")?;
     706            0 :         let pg_control_bytes = self
     707            0 :             .timeline
     708            0 :             .get_control_file(self.lsn, self.ctx)
     709            0 :             .await
     710            0 :             .context("failed get control bytes")?;
     711              : 
     712            0 :         let (pg_control_bytes, system_identifier) = postgres_ffi::generate_pg_control(
     713            0 :             &pg_control_bytes,
     714            0 :             &checkpoint_bytes,
     715            0 :             self.lsn,
     716            0 :             self.timeline.pg_version,
     717            0 :         )?;
     718              : 
     719              :         //send pg_control
     720            0 :         let header = new_tar_header("global/pg_control", pg_control_bytes.len() as u64)?;
     721            0 :         self.ar
     722            0 :             .append(&header, &pg_control_bytes[..])
     723            0 :             .await
     724            0 :             .map_err(|e| BasebackupError::Client(e, "add_pgcontrol_file,pg_control"))?;
     725              : 
     726              :         //send wal segment
     727            0 :         let segno = self.lsn.segment_number(WAL_SEGMENT_SIZE);
     728            0 :         let wal_file_name = XLogFileName(PG_TLI, segno, WAL_SEGMENT_SIZE);
     729            0 :         let wal_file_path = format!("pg_wal/{}", wal_file_name);
     730            0 :         let header = new_tar_header(&wal_file_path, WAL_SEGMENT_SIZE as u64)?;
     731              : 
     732            0 :         let wal_seg = postgres_ffi::generate_wal_segment(
     733            0 :             segno,
     734            0 :             system_identifier,
     735            0 :             self.timeline.pg_version,
     736            0 :             self.lsn,
     737            0 :         )
     738            0 :         .map_err(|e| anyhow!(e).context("Failed generating wal segment"))?;
     739            0 :         if wal_seg.len() != WAL_SEGMENT_SIZE {
     740            0 :             return Err(BasebackupError::Server(anyhow!(
     741            0 :                 "wal_seg.len() != WAL_SEGMENT_SIZE, wal_seg.len()={}",
     742            0 :                 wal_seg.len()
     743            0 :             )));
     744            0 :         }
     745            0 :         self.ar
     746            0 :             .append(&header, &wal_seg[..])
     747            0 :             .await
     748            0 :             .map_err(|e| BasebackupError::Client(e, "add_pgcontrol_file,wal_segment"))?;
     749            0 :         Ok(())
     750            0 :     }
     751              : }
     752              : 
     753              : //
     754              : // Create new tarball entry header
     755              : //
     756            0 : fn new_tar_header(path: &str, size: u64) -> anyhow::Result<Header> {
     757            0 :     let mut header = Header::new_gnu();
     758            0 :     header.set_size(size);
     759            0 :     header.set_path(path)?;
     760            0 :     header.set_mode(0b110000000); // -rw-------
     761            0 :     header.set_mtime(
     762            0 :         // use currenttime as last modified time
     763            0 :         SystemTime::now()
     764            0 :             .duration_since(SystemTime::UNIX_EPOCH)
     765            0 :             .unwrap()
     766            0 :             .as_secs(),
     767            0 :     );
     768            0 :     header.set_cksum();
     769            0 :     Ok(header)
     770            0 : }
     771              : 
     772            0 : fn new_tar_header_dir(path: &str) -> anyhow::Result<Header> {
     773            0 :     let mut header = Header::new_gnu();
     774            0 :     header.set_size(0);
     775            0 :     header.set_path(path)?;
     776            0 :     header.set_mode(0o755); // -rw-------
     777            0 :     header.set_entry_type(EntryType::dir());
     778            0 :     header.set_mtime(
     779            0 :         // use currenttime as last modified time
     780            0 :         SystemTime::now()
     781            0 :             .duration_since(SystemTime::UNIX_EPOCH)
     782            0 :             .unwrap()
     783            0 :             .as_secs(),
     784            0 :     );
     785            0 :     header.set_cksum();
     786            0 :     Ok(header)
     787            0 : }
        

Generated by: LCOV version 2.1-beta