LCOV - code coverage report
Current view: top level - pageserver/src - walingest.rs (source / functions) Coverage Total Hit
Test: 6df3fc19ec669bcfbbf9aba41d1338898d24eaa0.info Lines: 59.8 % 1816 1086
Test Date: 2025-03-12 18:28:53 Functions: 58.2 % 79 46

            Line data    Source code
       1              : //!
       2              : //! Parse PostgreSQL WAL records and store them in a neon Timeline.
       3              : //!
       4              : //! The pipeline for ingesting WAL looks like this:
       5              : //!
       6              : //! WAL receiver  -> [`wal_decoder`] ->  WalIngest  ->   Repository
       7              : //!
       8              : //! The WAL receiver receives a stream of WAL from the WAL safekeepers.
       9              : //! Records get decoded and interpreted in the [`wal_decoder`] module
      10              : //! and then stored to the Repository by WalIngest.
      11              : //!
      12              : //! The neon Repository can store page versions in two formats: as
      13              : //! page images, or a WAL records. [`wal_decoder::models::InterpretedWalRecord::from_bytes_filtered`]
      14              : //! extracts page images out of some WAL records, but mostly it's WAL
      15              : //! records. If a WAL record modifies multiple pages, WalIngest
      16              : //! will call Repository::put_rel_wal_record or put_rel_page_image functions
      17              : //! separately for each modified page.
      18              : //!
      19              : //! To reconstruct a page using a WAL record, the Repository calls the
      20              : //! code in walredo.rs. walredo.rs passes most WAL records to the WAL
      21              : //! redo Postgres process, but some records it can handle directly with
      22              : //! bespoken Rust code.
      23              : 
      24              : use std::collections::HashMap;
      25              : use std::sync::{Arc, OnceLock};
      26              : use std::time::{Duration, Instant, SystemTime};
      27              : 
      28              : use anyhow::{Result, bail};
      29              : use bytes::{Buf, Bytes};
      30              : use pageserver_api::key::rel_block_to_key;
      31              : use pageserver_api::record::NeonWalRecord;
      32              : use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
      33              : use pageserver_api::shard::ShardIdentity;
      34              : use postgres_ffi::relfile_utils::{FSM_FORKNUM, INIT_FORKNUM, MAIN_FORKNUM, VISIBILITYMAP_FORKNUM};
      35              : use postgres_ffi::walrecord::*;
      36              : use postgres_ffi::{
      37              :     TimestampTz, TransactionId, dispatch_pgversion, enum_pgversion, enum_pgversion_dispatch,
      38              :     fsm_logical_to_physical, pg_constants,
      39              : };
      40              : use tracing::*;
      41              : use utils::bin_ser::SerializeError;
      42              : use utils::lsn::Lsn;
      43              : use utils::rate_limit::RateLimit;
      44              : use utils::{critical, failpoint_support};
      45              : use wal_decoder::models::*;
      46              : 
      47              : use crate::ZERO_PAGE;
      48              : use crate::context::RequestContext;
      49              : use crate::metrics::WAL_INGEST;
      50              : use crate::pgdatadir_mapping::{DatadirModification, Version};
      51              : use crate::span::debug_assert_current_span_has_tenant_and_timeline_id;
      52              : use crate::tenant::{PageReconstructError, Timeline};
      53              : 
      54              : enum_pgversion! {CheckPoint, pgv::CheckPoint}
      55              : 
      56              : impl CheckPoint {
      57           12 :     fn encode(&self) -> Result<Bytes, SerializeError> {
      58           12 :         enum_pgversion_dispatch!(self, CheckPoint, cp, { cp.encode() })
      59           12 :     }
      60              : 
      61       291668 :     fn update_next_xid(&mut self, xid: u32) -> bool {
      62       291668 :         enum_pgversion_dispatch!(self, CheckPoint, cp, { cp.update_next_xid(xid) })
      63       291668 :     }
      64              : 
      65            0 :     pub fn update_next_multixid(&mut self, multi_xid: u32, multi_offset: u32) -> bool {
      66            0 :         enum_pgversion_dispatch!(self, CheckPoint, cp, {
      67            0 :             cp.update_next_multixid(multi_xid, multi_offset)
      68              :         })
      69            0 :     }
      70              : }
      71              : 
      72              : /// Temporary limitation of WAL lag warnings after attach
      73              : ///
      74              : /// After tenant attach, we want to limit WAL lag warnings because
      75              : /// we don't look at the WAL until the attach is complete, which
      76              : /// might take a while.
      77              : pub struct WalLagCooldown {
      78              :     /// Until when should this limitation apply at all
      79              :     active_until: std::time::Instant,
      80              :     /// The maximum lag to suppress. Lags above this limit get reported anyways.
      81              :     max_lag: Duration,
      82              : }
      83              : 
      84              : impl WalLagCooldown {
      85            0 :     pub fn new(attach_start: Instant, attach_duration: Duration) -> Self {
      86            0 :         Self {
      87            0 :             active_until: attach_start + attach_duration * 3 + Duration::from_secs(120),
      88            0 :             max_lag: attach_duration * 2 + Duration::from_secs(60),
      89            0 :         }
      90            0 :     }
      91              : }
      92              : 
      93              : pub struct WalIngest {
      94              :     attach_wal_lag_cooldown: Arc<OnceLock<WalLagCooldown>>,
      95              :     shard: ShardIdentity,
      96              :     checkpoint: CheckPoint,
      97              :     checkpoint_modified: bool,
      98              :     warn_ingest_lag: WarnIngestLag,
      99              : }
     100              : 
     101              : struct WarnIngestLag {
     102              :     lag_msg_ratelimit: RateLimit,
     103              :     future_lsn_msg_ratelimit: RateLimit,
     104              :     timestamp_invalid_msg_ratelimit: RateLimit,
     105              : }
     106              : 
     107              : impl WalIngest {
     108           24 :     pub async fn new(
     109           24 :         timeline: &Timeline,
     110           24 :         startpoint: Lsn,
     111           24 :         ctx: &RequestContext,
     112           24 :     ) -> anyhow::Result<WalIngest> {
     113              :         // Fetch the latest checkpoint into memory, so that we can compare with it
     114              :         // quickly in `ingest_record` and update it when it changes.
     115           24 :         let checkpoint_bytes = timeline.get_checkpoint(startpoint, ctx).await?;
     116           24 :         let pgversion = timeline.pg_version;
     117              : 
     118           24 :         let checkpoint = dispatch_pgversion!(pgversion, {
     119            0 :             let checkpoint = pgv::CheckPoint::decode(&checkpoint_bytes)?;
     120            0 :             trace!("CheckPoint.nextXid = {}", checkpoint.nextXid.value);
     121            0 :             <pgv::CheckPoint as Into<CheckPoint>>::into(checkpoint)
     122              :         });
     123              : 
     124           24 :         Ok(WalIngest {
     125           24 :             shard: *timeline.get_shard_identity(),
     126           24 :             checkpoint,
     127           24 :             checkpoint_modified: false,
     128           24 :             attach_wal_lag_cooldown: timeline.attach_wal_lag_cooldown.clone(),
     129           24 :             warn_ingest_lag: WarnIngestLag {
     130           24 :                 lag_msg_ratelimit: RateLimit::new(std::time::Duration::from_secs(10)),
     131           24 :                 future_lsn_msg_ratelimit: RateLimit::new(std::time::Duration::from_secs(10)),
     132           24 :                 timestamp_invalid_msg_ratelimit: RateLimit::new(std::time::Duration::from_secs(10)),
     133           24 :             },
     134           24 :         })
     135           24 :     }
     136              : 
     137              :     /// Ingest an interpreted PostgreSQL WAL record by doing writes to the underlying key value
     138              :     /// storage of a given timeline.
     139              :     ///
     140              :     /// This function updates `lsn` field of `DatadirModification`
     141              :     ///
     142              :     /// This function returns `true` if the record was ingested, and `false` if it was filtered out
     143       291704 :     pub async fn ingest_record(
     144       291704 :         &mut self,
     145       291704 :         interpreted: InterpretedWalRecord,
     146       291704 :         modification: &mut DatadirModification<'_>,
     147       291704 :         ctx: &RequestContext,
     148       291704 :     ) -> anyhow::Result<bool> {
     149       291704 :         WAL_INGEST.records_received.inc();
     150       291704 :         let prev_len = modification.len();
     151       291704 : 
     152       291704 :         modification.set_lsn(interpreted.next_record_lsn)?;
     153              : 
     154       291704 :         if matches!(interpreted.flush_uncommitted, FlushUncommittedRecords::Yes) {
     155              :             // Records of this type should always be preceded by a commit(), as they
     156              :             // rely on reading data pages back from the Timeline.
     157            0 :             assert!(!modification.has_dirty_data());
     158       291704 :         }
     159              : 
     160       291704 :         assert!(!self.checkpoint_modified);
     161       291704 :         if interpreted.xid != pg_constants::INVALID_TRANSACTION_ID
     162       291668 :             && self.checkpoint.update_next_xid(interpreted.xid)
     163            4 :         {
     164            4 :             self.checkpoint_modified = true;
     165       291700 :         }
     166              : 
     167       291704 :         failpoint_support::sleep_millis_async!("wal-ingest-record-sleep");
     168              : 
     169          132 :         match interpreted.metadata_record {
     170           24 :             Some(MetadataRecord::Heapam(rec)) => match rec {
     171           24 :                 HeapamRecord::ClearVmBits(clear_vm_bits) => {
     172           24 :                     self.ingest_clear_vm_bits(clear_vm_bits, modification, ctx)
     173           24 :                         .await?;
     174              :                 }
     175              :             },
     176            0 :             Some(MetadataRecord::Neonrmgr(rec)) => match rec {
     177            0 :                 NeonrmgrRecord::ClearVmBits(clear_vm_bits) => {
     178            0 :                     self.ingest_clear_vm_bits(clear_vm_bits, modification, ctx)
     179            0 :                         .await?;
     180              :                 }
     181              :             },
     182           32 :             Some(MetadataRecord::Smgr(rec)) => match rec {
     183           32 :                 SmgrRecord::Create(create) => {
     184           32 :                     self.ingest_xlog_smgr_create(create, modification, ctx)
     185           32 :                         .await?;
     186              :                 }
     187            0 :                 SmgrRecord::Truncate(truncate) => {
     188            0 :                     self.ingest_xlog_smgr_truncate(truncate, modification, ctx)
     189            0 :                         .await?;
     190              :                 }
     191              :             },
     192            0 :             Some(MetadataRecord::Dbase(rec)) => match rec {
     193            0 :                 DbaseRecord::Create(create) => {
     194            0 :                     self.ingest_xlog_dbase_create(create, modification, ctx)
     195            0 :                         .await?;
     196              :                 }
     197            0 :                 DbaseRecord::Drop(drop) => {
     198            0 :                     self.ingest_xlog_dbase_drop(drop, modification, ctx).await?;
     199              :                 }
     200              :             },
     201            0 :             Some(MetadataRecord::Clog(rec)) => match rec {
     202            0 :                 ClogRecord::ZeroPage(zero_page) => {
     203            0 :                     self.ingest_clog_zero_page(zero_page, modification, ctx)
     204            0 :                         .await?;
     205              :                 }
     206            0 :                 ClogRecord::Truncate(truncate) => {
     207            0 :                     self.ingest_clog_truncate(truncate, modification, ctx)
     208            0 :                         .await?;
     209              :                 }
     210              :             },
     211           16 :             Some(MetadataRecord::Xact(rec)) => {
     212           16 :                 self.ingest_xact_record(rec, modification, ctx).await?;
     213              :             }
     214            0 :             Some(MetadataRecord::MultiXact(rec)) => match rec {
     215            0 :                 MultiXactRecord::ZeroPage(zero_page) => {
     216            0 :                     self.ingest_multixact_zero_page(zero_page, modification, ctx)
     217            0 :                         .await?;
     218              :                 }
     219            0 :                 MultiXactRecord::Create(create) => {
     220            0 :                     self.ingest_multixact_create(modification, &create)?;
     221              :                 }
     222            0 :                 MultiXactRecord::Truncate(truncate) => {
     223            0 :                     self.ingest_multixact_truncate(modification, &truncate, ctx)
     224            0 :                         .await?;
     225              :                 }
     226              :             },
     227            0 :             Some(MetadataRecord::Relmap(rec)) => match rec {
     228            0 :                 RelmapRecord::Update(update) => {
     229            0 :                     self.ingest_relmap_update(update, modification, ctx).await?;
     230              :                 }
     231              :             },
     232           60 :             Some(MetadataRecord::Xlog(rec)) => match rec {
     233           60 :                 XlogRecord::Raw(raw) => {
     234           60 :                     self.ingest_raw_xlog_record(raw, modification, ctx).await?;
     235              :                 }
     236              :             },
     237            0 :             Some(MetadataRecord::LogicalMessage(rec)) => match rec {
     238            0 :                 LogicalMessageRecord::Put(put) => {
     239            0 :                     self.ingest_logical_message_put(put, modification, ctx)
     240            0 :                         .await?;
     241              :                 }
     242              :                 #[cfg(feature = "testing")]
     243              :                 LogicalMessageRecord::Failpoint => {
     244              :                     // This is a convenient way to make the WAL ingestion pause at
     245              :                     // particular point in the WAL. For more fine-grained control,
     246              :                     // we could peek into the message and only pause if it contains
     247              :                     // a particular string, for example, but this is enough for now.
     248            0 :                     failpoint_support::sleep_millis_async!(
     249            0 :                         "pageserver-wal-ingest-logical-message-sleep"
     250            0 :                     );
     251              :                 }
     252              :             },
     253            0 :             Some(MetadataRecord::Standby(rec)) => {
     254            0 :                 self.ingest_standby_record(rec).unwrap();
     255            0 :             }
     256            0 :             Some(MetadataRecord::Replorigin(rec)) => {
     257            0 :                 self.ingest_replorigin_record(rec, modification).await?;
     258              :             }
     259       291572 :             None => {
     260       291572 :                 // There are two cases through which we end up here:
     261       291572 :                 // 1. The resource manager for the original PG WAL record
     262       291572 :                 //    is [`pg_constants::RM_TBLSPC_ID`]. This is not a supported
     263       291572 :                 //    record type within Neon.
     264       291572 :                 // 2. The resource manager id was unknown to
     265       291572 :                 //    [`wal_decoder::decoder::MetadataRecord::from_decoded`].
     266       291572 :                 // TODO(vlad): Tighten this up more once we build confidence
     267       291572 :                 // that case (2) does not happen in the field.
     268       291572 :             }
     269              :         }
     270              : 
     271       291704 :         modification
     272       291704 :             .ingest_batch(interpreted.batch, &self.shard, ctx)
     273       291704 :             .await?;
     274              : 
     275              :         // If checkpoint data was updated, store the new version in the repository
     276       291704 :         if self.checkpoint_modified {
     277           12 :             let new_checkpoint_bytes = self.checkpoint.encode()?;
     278              : 
     279           12 :             modification.put_checkpoint(new_checkpoint_bytes)?;
     280           12 :             self.checkpoint_modified = false;
     281       291692 :         }
     282              : 
     283              :         // Note that at this point this record is only cached in the modification
     284              :         // until commit() is called to flush the data into the repository and update
     285              :         // the latest LSN.
     286              : 
     287       291704 :         Ok(modification.len() > prev_len)
     288       291704 :     }
     289              : 
     290              :     /// This is the same as AdjustToFullTransactionId(xid) in PostgreSQL
     291            0 :     fn adjust_to_full_transaction_id(&self, xid: TransactionId) -> Result<u64> {
     292            0 :         let next_full_xid =
     293            0 :             enum_pgversion_dispatch!(&self.checkpoint, CheckPoint, cp, { cp.nextXid.value });
     294              : 
     295            0 :         let next_xid = (next_full_xid) as u32;
     296            0 :         let mut epoch = (next_full_xid >> 32) as u32;
     297            0 : 
     298            0 :         if xid > next_xid {
     299              :             // Wraparound occurred, must be from a prev epoch.
     300            0 :             if epoch == 0 {
     301            0 :                 bail!(
     302            0 :                     "apparent XID wraparound with prepared transaction XID {xid}, nextXid is {next_full_xid}"
     303            0 :                 );
     304            0 :             }
     305            0 :             epoch -= 1;
     306            0 :         }
     307              : 
     308            0 :         Ok(((epoch as u64) << 32) | xid as u64)
     309            0 :     }
     310              : 
     311           24 :     async fn ingest_clear_vm_bits(
     312           24 :         &mut self,
     313           24 :         clear_vm_bits: ClearVmBits,
     314           24 :         modification: &mut DatadirModification<'_>,
     315           24 :         ctx: &RequestContext,
     316           24 :     ) -> anyhow::Result<()> {
     317           24 :         let ClearVmBits {
     318           24 :             new_heap_blkno,
     319           24 :             old_heap_blkno,
     320           24 :             flags,
     321           24 :             vm_rel,
     322           24 :         } = clear_vm_bits;
     323           24 :         // Clear the VM bits if required.
     324           24 :         let mut new_vm_blk = new_heap_blkno.map(pg_constants::HEAPBLK_TO_MAPBLOCK);
     325           24 :         let mut old_vm_blk = old_heap_blkno.map(pg_constants::HEAPBLK_TO_MAPBLOCK);
     326              : 
     327              :         // VM bits can only be cleared on the shard(s) owning the VM relation, and must be within
     328              :         // its view of the VM relation size. Out of caution, error instead of failing WAL ingestion,
     329              :         // as there has historically been cases where PostgreSQL has cleared spurious VM pages. See:
     330              :         // https://github.com/neondatabase/neon/pull/10634.
     331           24 :         let Some(vm_size) = get_relsize(modification, vm_rel, ctx).await? else {
     332            0 :             critical!("clear_vm_bits for unknown VM relation {vm_rel}");
     333            0 :             return Ok(());
     334              :         };
     335           24 :         if let Some(blknum) = new_vm_blk {
     336           24 :             if blknum >= vm_size {
     337            0 :                 critical!("new_vm_blk {blknum} not in {vm_rel} of size {vm_size}");
     338            0 :                 new_vm_blk = None;
     339           24 :             }
     340            0 :         }
     341           24 :         if let Some(blknum) = old_vm_blk {
     342            0 :             if blknum >= vm_size {
     343            0 :                 critical!("old_vm_blk {blknum} not in {vm_rel} of size {vm_size}");
     344            0 :                 old_vm_blk = None;
     345            0 :             }
     346           24 :         }
     347              : 
     348           24 :         if new_vm_blk.is_none() && old_vm_blk.is_none() {
     349            0 :             return Ok(());
     350           24 :         } else if new_vm_blk == old_vm_blk {
     351              :             // An UPDATE record that needs to clear the bits for both old and the new page, both of
     352              :             // which reside on the same VM page.
     353            0 :             self.put_rel_wal_record(
     354            0 :                 modification,
     355            0 :                 vm_rel,
     356            0 :                 new_vm_blk.unwrap(),
     357            0 :                 NeonWalRecord::ClearVisibilityMapFlags {
     358            0 :                     new_heap_blkno,
     359            0 :                     old_heap_blkno,
     360            0 :                     flags,
     361            0 :                 },
     362            0 :                 ctx,
     363            0 :             )
     364            0 :             .await?;
     365              :         } else {
     366              :             // Clear VM bits for one heap page, or for two pages that reside on different VM pages.
     367           24 :             if let Some(new_vm_blk) = new_vm_blk {
     368           24 :                 self.put_rel_wal_record(
     369           24 :                     modification,
     370           24 :                     vm_rel,
     371           24 :                     new_vm_blk,
     372           24 :                     NeonWalRecord::ClearVisibilityMapFlags {
     373           24 :                         new_heap_blkno,
     374           24 :                         old_heap_blkno: None,
     375           24 :                         flags,
     376           24 :                     },
     377           24 :                     ctx,
     378           24 :                 )
     379           24 :                 .await?;
     380            0 :             }
     381           24 :             if let Some(old_vm_blk) = old_vm_blk {
     382            0 :                 self.put_rel_wal_record(
     383            0 :                     modification,
     384            0 :                     vm_rel,
     385            0 :                     old_vm_blk,
     386            0 :                     NeonWalRecord::ClearVisibilityMapFlags {
     387            0 :                         new_heap_blkno: None,
     388            0 :                         old_heap_blkno,
     389            0 :                         flags,
     390            0 :                     },
     391            0 :                     ctx,
     392            0 :                 )
     393            0 :                 .await?;
     394           24 :             }
     395              :         }
     396           24 :         Ok(())
     397           24 :     }
     398              : 
     399              :     /// Subroutine of ingest_record(), to handle an XLOG_DBASE_CREATE record.
     400            0 :     async fn ingest_xlog_dbase_create(
     401            0 :         &mut self,
     402            0 :         create: DbaseCreate,
     403            0 :         modification: &mut DatadirModification<'_>,
     404            0 :         ctx: &RequestContext,
     405            0 :     ) -> anyhow::Result<()> {
     406            0 :         let DbaseCreate {
     407            0 :             db_id,
     408            0 :             tablespace_id,
     409            0 :             src_db_id,
     410            0 :             src_tablespace_id,
     411            0 :         } = create;
     412              : 
     413            0 :         let rels = modification
     414            0 :             .tline
     415            0 :             .list_rels(
     416            0 :                 src_tablespace_id,
     417            0 :                 src_db_id,
     418            0 :                 Version::Modified(modification),
     419            0 :                 ctx,
     420            0 :             )
     421            0 :             .await?;
     422              : 
     423            0 :         debug!("ingest_xlog_dbase_create: {} rels", rels.len());
     424              : 
     425              :         // Copy relfilemap
     426            0 :         let filemap = modification
     427            0 :             .tline
     428            0 :             .get_relmap_file(
     429            0 :                 src_tablespace_id,
     430            0 :                 src_db_id,
     431            0 :                 Version::Modified(modification),
     432            0 :                 ctx,
     433            0 :             )
     434            0 :             .await?;
     435            0 :         modification
     436            0 :             .put_relmap_file(tablespace_id, db_id, filemap, ctx)
     437            0 :             .await?;
     438              : 
     439            0 :         let mut num_rels_copied = 0;
     440            0 :         let mut num_blocks_copied = 0;
     441            0 :         for src_rel in rels {
     442            0 :             assert_eq!(src_rel.spcnode, src_tablespace_id);
     443            0 :             assert_eq!(src_rel.dbnode, src_db_id);
     444              : 
     445            0 :             let nblocks = modification
     446            0 :                 .tline
     447            0 :                 .get_rel_size(src_rel, Version::Modified(modification), ctx)
     448            0 :                 .await?;
     449            0 :             let dst_rel = RelTag {
     450            0 :                 spcnode: tablespace_id,
     451            0 :                 dbnode: db_id,
     452            0 :                 relnode: src_rel.relnode,
     453            0 :                 forknum: src_rel.forknum,
     454            0 :             };
     455            0 : 
     456            0 :             modification.put_rel_creation(dst_rel, nblocks, ctx).await?;
     457              : 
     458              :             // Copy content
     459            0 :             debug!("copying rel {} to {}, {} blocks", src_rel, dst_rel, nblocks);
     460            0 :             for blknum in 0..nblocks {
     461              :                 // Sharding:
     462              :                 //  - src and dst are always on the same shard, because they differ only by dbNode, and
     463              :                 //    dbNode is not included in the hash inputs for sharding.
     464              :                 //  - This WAL command is replayed on all shards, but each shard only copies the blocks
     465              :                 //    that belong to it.
     466            0 :                 let src_key = rel_block_to_key(src_rel, blknum);
     467            0 :                 if !self.shard.is_key_local(&src_key) {
     468            0 :                     debug!(
     469            0 :                         "Skipping non-local key {} during XLOG_DBASE_CREATE",
     470              :                         src_key
     471              :                     );
     472            0 :                     continue;
     473            0 :                 }
     474            0 :                 debug!(
     475            0 :                     "copying block {} from {} ({}) to {}",
     476              :                     blknum, src_rel, src_key, dst_rel
     477              :                 );
     478              : 
     479            0 :                 let content = modification
     480            0 :                     .tline
     481            0 :                     .get_rel_page_at_lsn(
     482            0 :                         src_rel,
     483            0 :                         blknum,
     484            0 :                         Version::Modified(modification),
     485            0 :                         ctx,
     486            0 :                         crate::tenant::storage_layer::IoConcurrency::sequential(),
     487            0 :                     )
     488            0 :                     .await?;
     489            0 :                 modification.put_rel_page_image(dst_rel, blknum, content)?;
     490            0 :                 num_blocks_copied += 1;
     491              :             }
     492              : 
     493            0 :             num_rels_copied += 1;
     494              :         }
     495              : 
     496            0 :         info!(
     497            0 :             "Created database {}/{}, copied {} blocks in {} rels",
     498              :             tablespace_id, db_id, num_blocks_copied, num_rels_copied
     499              :         );
     500            0 :         Ok(())
     501            0 :     }
     502              : 
     503            0 :     async fn ingest_xlog_dbase_drop(
     504            0 :         &mut self,
     505            0 :         dbase_drop: DbaseDrop,
     506            0 :         modification: &mut DatadirModification<'_>,
     507            0 :         ctx: &RequestContext,
     508            0 :     ) -> anyhow::Result<()> {
     509            0 :         let DbaseDrop {
     510            0 :             db_id,
     511            0 :             tablespace_ids,
     512            0 :         } = dbase_drop;
     513            0 :         for tablespace_id in tablespace_ids {
     514            0 :             trace!("Drop db {}, {}", tablespace_id, db_id);
     515            0 :             modification.drop_dbdir(tablespace_id, db_id, ctx).await?;
     516              :         }
     517              : 
     518            0 :         Ok(())
     519            0 :     }
     520              : 
     521           32 :     async fn ingest_xlog_smgr_create(
     522           32 :         &mut self,
     523           32 :         create: SmgrCreate,
     524           32 :         modification: &mut DatadirModification<'_>,
     525           32 :         ctx: &RequestContext,
     526           32 :     ) -> anyhow::Result<()> {
     527           32 :         let SmgrCreate { rel } = create;
     528           32 :         self.put_rel_creation(modification, rel, ctx).await?;
     529           32 :         Ok(())
     530           32 :     }
     531              : 
     532              :     /// Subroutine of ingest_record(), to handle an XLOG_SMGR_TRUNCATE record.
     533              :     ///
     534              :     /// This is the same logic as in PostgreSQL's smgr_redo() function.
     535            0 :     async fn ingest_xlog_smgr_truncate(
     536            0 :         &mut self,
     537            0 :         truncate: XlSmgrTruncate,
     538            0 :         modification: &mut DatadirModification<'_>,
     539            0 :         ctx: &RequestContext,
     540            0 :     ) -> anyhow::Result<()> {
     541            0 :         let XlSmgrTruncate {
     542            0 :             blkno,
     543            0 :             rnode,
     544            0 :             flags,
     545            0 :         } = truncate;
     546            0 : 
     547            0 :         let spcnode = rnode.spcnode;
     548            0 :         let dbnode = rnode.dbnode;
     549            0 :         let relnode = rnode.relnode;
     550            0 : 
     551            0 :         if flags & pg_constants::SMGR_TRUNCATE_HEAP != 0 {
     552            0 :             let rel = RelTag {
     553            0 :                 spcnode,
     554            0 :                 dbnode,
     555            0 :                 relnode,
     556            0 :                 forknum: MAIN_FORKNUM,
     557            0 :             };
     558            0 : 
     559            0 :             self.put_rel_truncation(modification, rel, blkno, ctx)
     560            0 :                 .await?;
     561            0 :         }
     562            0 :         if flags & pg_constants::SMGR_TRUNCATE_FSM != 0 {
     563            0 :             let rel = RelTag {
     564            0 :                 spcnode,
     565            0 :                 dbnode,
     566            0 :                 relnode,
     567            0 :                 forknum: FSM_FORKNUM,
     568            0 :             };
     569            0 : 
     570            0 :             // Zero out the last remaining FSM page, if this shard owns it. We are not precise here,
     571            0 :             // and instead of digging in the FSM bitmap format we just clear the whole page.
     572            0 :             let fsm_logical_page_no = blkno / pg_constants::SLOTS_PER_FSM_PAGE;
     573            0 :             let mut fsm_physical_page_no = fsm_logical_to_physical(fsm_logical_page_no);
     574            0 :             if blkno % pg_constants::SLOTS_PER_FSM_PAGE != 0
     575            0 :                 && self
     576            0 :                     .shard
     577            0 :                     .is_key_local(&rel_block_to_key(rel, fsm_physical_page_no))
     578              :             {
     579            0 :                 modification.put_rel_page_image_zero(rel, fsm_physical_page_no)?;
     580            0 :                 fsm_physical_page_no += 1;
     581            0 :             }
     582              :             // Truncate this shard's view of the FSM relation size, if it even has one.
     583            0 :             let nblocks = get_relsize(modification, rel, ctx).await?.unwrap_or(0);
     584            0 :             if nblocks > fsm_physical_page_no {
     585            0 :                 self.put_rel_truncation(modification, rel, fsm_physical_page_no, ctx)
     586            0 :                     .await?;
     587            0 :             }
     588            0 :         }
     589            0 :         if flags & pg_constants::SMGR_TRUNCATE_VM != 0 {
     590            0 :             let rel = RelTag {
     591            0 :                 spcnode,
     592            0 :                 dbnode,
     593            0 :                 relnode,
     594            0 :                 forknum: VISIBILITYMAP_FORKNUM,
     595            0 :             };
     596            0 : 
     597            0 :             // last remaining block, byte, and bit
     598            0 :             let mut vm_page_no = blkno / (pg_constants::VM_HEAPBLOCKS_PER_PAGE as u32);
     599            0 :             let trunc_byte = blkno as usize % pg_constants::VM_HEAPBLOCKS_PER_PAGE
     600            0 :                 / pg_constants::VM_HEAPBLOCKS_PER_BYTE;
     601            0 :             let trunc_offs = blkno as usize % pg_constants::VM_HEAPBLOCKS_PER_BYTE
     602            0 :                 * pg_constants::VM_BITS_PER_HEAPBLOCK;
     603            0 : 
     604            0 :             // Unless the new size is exactly at a visibility map page boundary, the
     605            0 :             // tail bits in the last remaining map page, representing truncated heap
     606            0 :             // blocks, need to be cleared. This is not only tidy, but also necessary
     607            0 :             // because we don't get a chance to clear the bits if the heap is extended
     608            0 :             // again. Only do this on the shard that owns the page.
     609            0 :             if (trunc_byte != 0 || trunc_offs != 0)
     610            0 :                 && self.shard.is_key_local(&rel_block_to_key(rel, vm_page_no))
     611              :             {
     612            0 :                 modification.put_rel_wal_record(
     613            0 :                     rel,
     614            0 :                     vm_page_no,
     615            0 :                     NeonWalRecord::TruncateVisibilityMap {
     616            0 :                         trunc_byte,
     617            0 :                         trunc_offs,
     618            0 :                     },
     619            0 :                 )?;
     620            0 :                 vm_page_no += 1;
     621            0 :             }
     622              :             // Truncate this shard's view of the VM relation size, if it even has one.
     623            0 :             let nblocks = get_relsize(modification, rel, ctx).await?.unwrap_or(0);
     624            0 :             if nblocks > vm_page_no {
     625            0 :                 self.put_rel_truncation(modification, rel, vm_page_no, ctx)
     626            0 :                     .await?;
     627            0 :             }
     628            0 :         }
     629            0 :         Ok(())
     630            0 :     }
     631              : 
     632           16 :     fn warn_on_ingest_lag(
     633           16 :         &mut self,
     634           16 :         conf: &crate::config::PageServerConf,
     635           16 :         wal_timestamp: TimestampTz,
     636           16 :     ) {
     637           16 :         debug_assert_current_span_has_tenant_and_timeline_id();
     638           16 :         let now = SystemTime::now();
     639           16 :         let rate_limits = &mut self.warn_ingest_lag;
     640              : 
     641           16 :         let ts = enum_pgversion_dispatch!(&self.checkpoint, CheckPoint, _cp, {
     642            0 :             pgv::xlog_utils::try_from_pg_timestamp(wal_timestamp)
     643              :         });
     644              : 
     645           16 :         match ts {
     646           16 :             Ok(ts) => {
     647           16 :                 match now.duration_since(ts) {
     648           16 :                     Ok(lag) => {
     649           16 :                         if lag > conf.wait_lsn_timeout {
     650           16 :                             rate_limits.lag_msg_ratelimit.call2(|rate_limit_stats| {
     651            4 :                                 if let Some(cooldown) = self.attach_wal_lag_cooldown.get() {
     652            0 :                                     if std::time::Instant::now() < cooldown.active_until && lag <= cooldown.max_lag {
     653            0 :                                         return;
     654            0 :                                     }
     655            4 :                                 } else {
     656            4 :                                     // Still loading? We shouldn't be here
     657            4 :                                 }
     658            4 :                                 let lag = humantime::format_duration(lag);
     659            4 :                                 warn!(%rate_limit_stats, %lag, "ingesting record with timestamp lagging more than wait_lsn_timeout");
     660           16 :                             })
     661            0 :                         }
     662              :                     }
     663            0 :                     Err(e) => {
     664            0 :                         let delta_t = e.duration();
     665              :                         // determined by prod victoriametrics query: 1000 * (timestamp(node_time_seconds{neon_service="pageserver"}) - node_time_seconds)
     666              :                         // => https://www.robustperception.io/time-metric-from-the-node-exporter/
     667              :                         const IGNORED_DRIFT: Duration = Duration::from_millis(100);
     668            0 :                         if delta_t > IGNORED_DRIFT {
     669            0 :                             let delta_t = humantime::format_duration(delta_t);
     670            0 :                             rate_limits.future_lsn_msg_ratelimit.call2(|rate_limit_stats| {
     671            0 :                                 warn!(%rate_limit_stats, %delta_t, "ingesting record with timestamp from future");
     672            0 :                             })
     673            0 :                         }
     674              :                     }
     675              :                 };
     676              :             }
     677            0 :             Err(error) => {
     678            0 :                 rate_limits.timestamp_invalid_msg_ratelimit.call2(|rate_limit_stats| {
     679            0 :                     warn!(%rate_limit_stats, %error, "ingesting record with invalid timestamp, cannot calculate lag and will fail find-lsn-for-timestamp type queries");
     680            0 :                 })
     681              :             }
     682              :         }
     683           16 :     }
     684              : 
     685              :     /// Subroutine of ingest_record(), to handle an XLOG_XACT_* records.
     686              :     ///
     687           16 :     async fn ingest_xact_record(
     688           16 :         &mut self,
     689           16 :         record: XactRecord,
     690           16 :         modification: &mut DatadirModification<'_>,
     691           16 :         ctx: &RequestContext,
     692           16 :     ) -> anyhow::Result<()> {
     693           16 :         let (xact_common, is_commit, is_prepared) = match record {
     694            0 :             XactRecord::Prepare(XactPrepare { xl_xid, data }) => {
     695            0 :                 let xid: u64 = if modification.tline.pg_version >= 17 {
     696            0 :                     self.adjust_to_full_transaction_id(xl_xid)?
     697              :                 } else {
     698            0 :                     xl_xid as u64
     699              :                 };
     700            0 :                 return modification.put_twophase_file(xid, data, ctx).await;
     701              :             }
     702           16 :             XactRecord::Commit(common) => (common, true, false),
     703            0 :             XactRecord::Abort(common) => (common, false, false),
     704            0 :             XactRecord::CommitPrepared(common) => (common, true, true),
     705            0 :             XactRecord::AbortPrepared(common) => (common, false, true),
     706              :         };
     707              : 
     708              :         let XactCommon {
     709           16 :             parsed,
     710           16 :             origin_id,
     711           16 :             xl_xid,
     712           16 :             lsn,
     713           16 :         } = xact_common;
     714           16 : 
     715           16 :         // Record update of CLOG pages
     716           16 :         let mut pageno = parsed.xid / pg_constants::CLOG_XACTS_PER_PAGE;
     717           16 :         let mut segno = pageno / pg_constants::SLRU_PAGES_PER_SEGMENT;
     718           16 :         let mut rpageno = pageno % pg_constants::SLRU_PAGES_PER_SEGMENT;
     719           16 :         let mut page_xids: Vec<TransactionId> = vec![parsed.xid];
     720           16 : 
     721           16 :         self.warn_on_ingest_lag(modification.tline.conf, parsed.xact_time);
     722              : 
     723           16 :         for subxact in &parsed.subxacts {
     724            0 :             let subxact_pageno = subxact / pg_constants::CLOG_XACTS_PER_PAGE;
     725            0 :             if subxact_pageno != pageno {
     726              :                 // This subxact goes to different page. Write the record
     727              :                 // for all the XIDs on the previous page, and continue
     728              :                 // accumulating XIDs on this new page.
     729            0 :                 modification.put_slru_wal_record(
     730            0 :                     SlruKind::Clog,
     731            0 :                     segno,
     732            0 :                     rpageno,
     733            0 :                     if is_commit {
     734            0 :                         NeonWalRecord::ClogSetCommitted {
     735            0 :                             xids: page_xids,
     736            0 :                             timestamp: parsed.xact_time,
     737            0 :                         }
     738              :                     } else {
     739            0 :                         NeonWalRecord::ClogSetAborted { xids: page_xids }
     740              :                     },
     741            0 :                 )?;
     742            0 :                 page_xids = Vec::new();
     743            0 :             }
     744            0 :             pageno = subxact_pageno;
     745            0 :             segno = pageno / pg_constants::SLRU_PAGES_PER_SEGMENT;
     746            0 :             rpageno = pageno % pg_constants::SLRU_PAGES_PER_SEGMENT;
     747            0 :             page_xids.push(*subxact);
     748              :         }
     749           16 :         modification.put_slru_wal_record(
     750           16 :             SlruKind::Clog,
     751           16 :             segno,
     752           16 :             rpageno,
     753           16 :             if is_commit {
     754           16 :                 NeonWalRecord::ClogSetCommitted {
     755           16 :                     xids: page_xids,
     756           16 :                     timestamp: parsed.xact_time,
     757           16 :                 }
     758              :             } else {
     759            0 :                 NeonWalRecord::ClogSetAborted { xids: page_xids }
     760              :             },
     761            0 :         )?;
     762              : 
     763              :         // Group relations to drop by dbNode.  This map will contain all relations that _might_
     764              :         // exist, we will reduce it to which ones really exist later.  This map can be huge if
     765              :         // the transaction touches a huge number of relations (there is no bound on this in
     766              :         // postgres).
     767           16 :         let mut drop_relations: HashMap<(u32, u32), Vec<RelTag>> = HashMap::new();
     768              : 
     769           16 :         for xnode in &parsed.xnodes {
     770            0 :             for forknum in MAIN_FORKNUM..=INIT_FORKNUM {
     771            0 :                 let rel = RelTag {
     772            0 :                     forknum,
     773            0 :                     spcnode: xnode.spcnode,
     774            0 :                     dbnode: xnode.dbnode,
     775            0 :                     relnode: xnode.relnode,
     776            0 :                 };
     777            0 :                 drop_relations
     778            0 :                     .entry((xnode.spcnode, xnode.dbnode))
     779            0 :                     .or_default()
     780            0 :                     .push(rel);
     781            0 :             }
     782              :         }
     783              : 
     784              :         // Execute relation drops in a batch: the number may be huge, so deleting individually is prohibitively expensive
     785           16 :         modification.put_rel_drops(drop_relations, ctx).await?;
     786              : 
     787           16 :         if origin_id != 0 {
     788            0 :             modification
     789            0 :                 .set_replorigin(origin_id, parsed.origin_lsn)
     790            0 :                 .await?;
     791           16 :         }
     792              : 
     793           16 :         if is_prepared {
     794              :             // Remove twophase file. see RemoveTwoPhaseFile() in postgres code
     795            0 :             trace!(
     796            0 :                 "Drop twophaseFile for xid {} parsed_xact.xid {} here at {}",
     797              :                 xl_xid, parsed.xid, lsn,
     798              :             );
     799              : 
     800            0 :             let xid: u64 = if modification.tline.pg_version >= 17 {
     801            0 :                 self.adjust_to_full_transaction_id(parsed.xid)?
     802              :             } else {
     803            0 :                 parsed.xid as u64
     804              :             };
     805            0 :             modification.drop_twophase_file(xid, ctx).await?;
     806           16 :         }
     807              : 
     808           16 :         Ok(())
     809           16 :     }
     810              : 
     811            0 :     async fn ingest_clog_truncate(
     812            0 :         &mut self,
     813            0 :         truncate: ClogTruncate,
     814            0 :         modification: &mut DatadirModification<'_>,
     815            0 :         ctx: &RequestContext,
     816            0 :     ) -> anyhow::Result<()> {
     817            0 :         let ClogTruncate {
     818            0 :             pageno,
     819            0 :             oldest_xid,
     820            0 :             oldest_xid_db,
     821            0 :         } = truncate;
     822            0 : 
     823            0 :         info!(
     824            0 :             "RM_CLOG_ID truncate pageno {} oldestXid {} oldestXidDB {}",
     825              :             pageno, oldest_xid, oldest_xid_db
     826              :         );
     827              : 
     828              :         // In Postgres, oldestXid and oldestXidDB are updated in memory when the CLOG is
     829              :         // truncated, but a checkpoint record with the updated values isn't written until
     830              :         // later. In Neon, a server can start at any LSN, not just on a checkpoint record,
     831              :         // so we keep the oldestXid and oldestXidDB up-to-date.
     832            0 :         enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, {
     833            0 :             cp.oldestXid = oldest_xid;
     834            0 :             cp.oldestXidDB = oldest_xid_db;
     835            0 :         });
     836            0 :         self.checkpoint_modified = true;
     837              : 
     838              :         // TODO Treat AdvanceOldestClogXid() or write a comment why we don't need it
     839              : 
     840            0 :         let latest_page_number =
     841            0 :             enum_pgversion_dispatch!(self.checkpoint, CheckPoint, cp, { cp.nextXid.value }) as u32
     842              :                 / pg_constants::CLOG_XACTS_PER_PAGE;
     843              : 
     844              :         // Now delete all segments containing pages between xlrec.pageno
     845              :         // and latest_page_number.
     846              : 
     847              :         // First, make an important safety check:
     848              :         // the current endpoint page must not be eligible for removal.
     849              :         // See SimpleLruTruncate() in slru.c
     850            0 :         if dispatch_pgversion!(modification.tline.pg_version, {
     851            0 :             pgv::nonrelfile_utils::clogpage_precedes(latest_page_number, pageno)
     852              :         }) {
     853            0 :             info!("could not truncate directory pg_xact apparent wraparound");
     854            0 :             return Ok(());
     855            0 :         }
     856            0 : 
     857            0 :         // Iterate via SLRU CLOG segments and drop segments that we're ready to truncate
     858            0 :         //
     859            0 :         // We cannot pass 'lsn' to the Timeline.list_nonrels(), or it
     860            0 :         // will block waiting for the last valid LSN to advance up to
     861            0 :         // it. So we use the previous record's LSN in the get calls
     862            0 :         // instead.
     863            0 :         if modification.tline.get_shard_identity().is_shard_zero() {
     864            0 :             for segno in modification
     865            0 :                 .tline
     866            0 :                 .list_slru_segments(SlruKind::Clog, Version::Modified(modification), ctx)
     867            0 :                 .await?
     868              :             {
     869            0 :                 let segpage = segno * pg_constants::SLRU_PAGES_PER_SEGMENT;
     870              : 
     871            0 :                 let may_delete = dispatch_pgversion!(modification.tline.pg_version, {
     872            0 :                     pgv::nonrelfile_utils::slru_may_delete_clogsegment(segpage, pageno)
     873              :                 });
     874              : 
     875            0 :                 if may_delete {
     876            0 :                     modification
     877            0 :                         .drop_slru_segment(SlruKind::Clog, segno, ctx)
     878            0 :                         .await?;
     879            0 :                     trace!("Drop CLOG segment {:>04X}", segno);
     880            0 :                 }
     881              :             }
     882            0 :         }
     883              : 
     884            0 :         Ok(())
     885            0 :     }
     886              : 
     887            0 :     async fn ingest_clog_zero_page(
     888            0 :         &mut self,
     889            0 :         zero_page: ClogZeroPage,
     890            0 :         modification: &mut DatadirModification<'_>,
     891            0 :         ctx: &RequestContext,
     892            0 :     ) -> anyhow::Result<()> {
     893            0 :         let ClogZeroPage { segno, rpageno } = zero_page;
     894            0 : 
     895            0 :         self.put_slru_page_image(
     896            0 :             modification,
     897            0 :             SlruKind::Clog,
     898            0 :             segno,
     899            0 :             rpageno,
     900            0 :             ZERO_PAGE.clone(),
     901            0 :             ctx,
     902            0 :         )
     903            0 :         .await
     904            0 :     }
     905              : 
     906            0 :     fn ingest_multixact_create(
     907            0 :         &mut self,
     908            0 :         modification: &mut DatadirModification,
     909            0 :         xlrec: &XlMultiXactCreate,
     910            0 :     ) -> Result<()> {
     911            0 :         // Create WAL record for updating the multixact-offsets page
     912            0 :         let pageno = xlrec.mid / pg_constants::MULTIXACT_OFFSETS_PER_PAGE as u32;
     913            0 :         let segno = pageno / pg_constants::SLRU_PAGES_PER_SEGMENT;
     914            0 :         let rpageno = pageno % pg_constants::SLRU_PAGES_PER_SEGMENT;
     915            0 : 
     916            0 :         modification.put_slru_wal_record(
     917            0 :             SlruKind::MultiXactOffsets,
     918            0 :             segno,
     919            0 :             rpageno,
     920            0 :             NeonWalRecord::MultixactOffsetCreate {
     921            0 :                 mid: xlrec.mid,
     922            0 :                 moff: xlrec.moff,
     923            0 :             },
     924            0 :         )?;
     925              : 
     926              :         // Create WAL records for the update of each affected multixact-members page
     927            0 :         let mut members = xlrec.members.iter();
     928            0 :         let mut offset = xlrec.moff;
     929              :         loop {
     930            0 :             let pageno = offset / pg_constants::MULTIXACT_MEMBERS_PER_PAGE as u32;
     931            0 : 
     932            0 :             // How many members fit on this page?
     933            0 :             let page_remain = pg_constants::MULTIXACT_MEMBERS_PER_PAGE as u32
     934            0 :                 - offset % pg_constants::MULTIXACT_MEMBERS_PER_PAGE as u32;
     935            0 : 
     936            0 :             let mut this_page_members: Vec<MultiXactMember> = Vec::new();
     937            0 :             for _ in 0..page_remain {
     938            0 :                 if let Some(m) = members.next() {
     939            0 :                     this_page_members.push(m.clone());
     940            0 :                 } else {
     941            0 :                     break;
     942              :                 }
     943              :             }
     944            0 :             if this_page_members.is_empty() {
     945              :                 // all done
     946            0 :                 break;
     947            0 :             }
     948            0 :             let n_this_page = this_page_members.len();
     949            0 : 
     950            0 :             modification.put_slru_wal_record(
     951            0 :                 SlruKind::MultiXactMembers,
     952            0 :                 pageno / pg_constants::SLRU_PAGES_PER_SEGMENT,
     953            0 :                 pageno % pg_constants::SLRU_PAGES_PER_SEGMENT,
     954            0 :                 NeonWalRecord::MultixactMembersCreate {
     955            0 :                     moff: offset,
     956            0 :                     members: this_page_members,
     957            0 :                 },
     958            0 :             )?;
     959              : 
     960              :             // Note: The multixact members can wrap around, even within one WAL record.
     961            0 :             offset = offset.wrapping_add(n_this_page as u32);
     962              :         }
     963            0 :         let next_offset = offset;
     964            0 :         assert!(xlrec.moff.wrapping_add(xlrec.nmembers) == next_offset);
     965              : 
     966              :         // Update next-multi-xid and next-offset
     967              :         //
     968              :         // NB: In PostgreSQL, the next-multi-xid stored in the control file is allowed to
     969              :         // go to 0, and it's fixed up by skipping to FirstMultiXactId in functions that
     970              :         // read it, like GetNewMultiXactId(). This is different from how nextXid is
     971              :         // incremented! nextXid skips over < FirstNormalTransactionId when the the value
     972              :         // is stored, so it's never 0 in a checkpoint.
     973              :         //
     974              :         // I don't know why it's done that way, it seems less error-prone to skip over 0
     975              :         // when the value is stored rather than when it's read. But let's do it the same
     976              :         // way here.
     977            0 :         let next_multi_xid = xlrec.mid.wrapping_add(1);
     978            0 : 
     979            0 :         if self
     980            0 :             .checkpoint
     981            0 :             .update_next_multixid(next_multi_xid, next_offset)
     982            0 :         {
     983            0 :             self.checkpoint_modified = true;
     984            0 :         }
     985              : 
     986              :         // Also update the next-xid with the highest member. According to the comments in
     987              :         // multixact_redo(), this shouldn't be necessary, but let's do the same here.
     988            0 :         let max_mbr_xid = xlrec.members.iter().fold(None, |acc, mbr| {
     989            0 :             if let Some(max_xid) = acc {
     990            0 :                 if mbr.xid.wrapping_sub(max_xid) as i32 > 0 {
     991            0 :                     Some(mbr.xid)
     992              :                 } else {
     993            0 :                     acc
     994              :                 }
     995              :             } else {
     996            0 :                 Some(mbr.xid)
     997              :             }
     998            0 :         });
     999              : 
    1000            0 :         if let Some(max_xid) = max_mbr_xid {
    1001            0 :             if self.checkpoint.update_next_xid(max_xid) {
    1002            0 :                 self.checkpoint_modified = true;
    1003            0 :             }
    1004            0 :         }
    1005            0 :         Ok(())
    1006            0 :     }
    1007              : 
    1008            0 :     async fn ingest_multixact_truncate(
    1009            0 :         &mut self,
    1010            0 :         modification: &mut DatadirModification<'_>,
    1011            0 :         xlrec: &XlMultiXactTruncate,
    1012            0 :         ctx: &RequestContext,
    1013            0 :     ) -> Result<()> {
    1014            0 :         let (maxsegment, startsegment, endsegment) =
    1015            0 :             enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, {
    1016            0 :                 cp.oldestMulti = xlrec.end_trunc_off;
    1017            0 :                 cp.oldestMultiDB = xlrec.oldest_multi_db;
    1018            0 :                 let maxsegment: i32 = pgv::nonrelfile_utils::mx_offset_to_member_segment(
    1019            0 :                     pg_constants::MAX_MULTIXACT_OFFSET,
    1020            0 :                 );
    1021            0 :                 let startsegment: i32 =
    1022            0 :                     pgv::nonrelfile_utils::mx_offset_to_member_segment(xlrec.start_trunc_memb);
    1023            0 :                 let endsegment: i32 =
    1024            0 :                     pgv::nonrelfile_utils::mx_offset_to_member_segment(xlrec.end_trunc_memb);
    1025            0 :                 (maxsegment, startsegment, endsegment)
    1026              :             });
    1027              : 
    1028            0 :         self.checkpoint_modified = true;
    1029            0 : 
    1030            0 :         // PerformMembersTruncation
    1031            0 :         let mut segment: i32 = startsegment;
    1032            0 : 
    1033            0 :         // Delete all the segments except the last one. The last segment can still
    1034            0 :         // contain, possibly partially, valid data.
    1035            0 :         if modification.tline.get_shard_identity().is_shard_zero() {
    1036            0 :             while segment != endsegment {
    1037            0 :                 modification
    1038            0 :                     .drop_slru_segment(SlruKind::MultiXactMembers, segment as u32, ctx)
    1039            0 :                     .await?;
    1040              : 
    1041              :                 /* move to next segment, handling wraparound correctly */
    1042            0 :                 if segment == maxsegment {
    1043            0 :                     segment = 0;
    1044            0 :                 } else {
    1045            0 :                     segment += 1;
    1046            0 :                 }
    1047              :             }
    1048            0 :         }
    1049              : 
    1050              :         // Truncate offsets
    1051              :         // FIXME: this did not handle wraparound correctly
    1052              : 
    1053            0 :         Ok(())
    1054            0 :     }
    1055              : 
    1056            0 :     async fn ingest_multixact_zero_page(
    1057            0 :         &mut self,
    1058            0 :         zero_page: MultiXactZeroPage,
    1059            0 :         modification: &mut DatadirModification<'_>,
    1060            0 :         ctx: &RequestContext,
    1061            0 :     ) -> Result<()> {
    1062            0 :         let MultiXactZeroPage {
    1063            0 :             slru_kind,
    1064            0 :             segno,
    1065            0 :             rpageno,
    1066            0 :         } = zero_page;
    1067            0 :         self.put_slru_page_image(
    1068            0 :             modification,
    1069            0 :             slru_kind,
    1070            0 :             segno,
    1071            0 :             rpageno,
    1072            0 :             ZERO_PAGE.clone(),
    1073            0 :             ctx,
    1074            0 :         )
    1075            0 :         .await
    1076            0 :     }
    1077              : 
    1078            0 :     async fn ingest_relmap_update(
    1079            0 :         &mut self,
    1080            0 :         update: RelmapUpdate,
    1081            0 :         modification: &mut DatadirModification<'_>,
    1082            0 :         ctx: &RequestContext,
    1083            0 :     ) -> Result<()> {
    1084            0 :         let RelmapUpdate { update, buf } = update;
    1085            0 : 
    1086            0 :         modification
    1087            0 :             .put_relmap_file(update.tsid, update.dbid, buf, ctx)
    1088            0 :             .await
    1089            0 :     }
    1090              : 
    1091           60 :     async fn ingest_raw_xlog_record(
    1092           60 :         &mut self,
    1093           60 :         raw_record: RawXlogRecord,
    1094           60 :         modification: &mut DatadirModification<'_>,
    1095           60 :         ctx: &RequestContext,
    1096           60 :     ) -> Result<()> {
    1097           60 :         let RawXlogRecord { info, lsn, mut buf } = raw_record;
    1098           60 :         let pg_version = modification.tline.pg_version;
    1099           60 : 
    1100           60 :         if info == pg_constants::XLOG_PARAMETER_CHANGE {
    1101            4 :             if let CheckPoint::V17(cp) = &mut self.checkpoint {
    1102            0 :                 let rec = v17::XlParameterChange::decode(&mut buf);
    1103            0 :                 cp.wal_level = rec.wal_level;
    1104            0 :                 self.checkpoint_modified = true;
    1105            4 :             }
    1106           56 :         } else if info == pg_constants::XLOG_END_OF_RECOVERY {
    1107            0 :             if let CheckPoint::V17(cp) = &mut self.checkpoint {
    1108            0 :                 let rec = v17::XlEndOfRecovery::decode(&mut buf);
    1109            0 :                 cp.wal_level = rec.wal_level;
    1110            0 :                 self.checkpoint_modified = true;
    1111            0 :             }
    1112           56 :         }
    1113              : 
    1114           60 :         enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, {
    1115            0 :             if info == pg_constants::XLOG_NEXTOID {
    1116            0 :                 let next_oid = buf.get_u32_le();
    1117            0 :                 if cp.nextOid != next_oid {
    1118            0 :                     cp.nextOid = next_oid;
    1119            0 :                     self.checkpoint_modified = true;
    1120            0 :                 }
    1121            0 :             } else if info == pg_constants::XLOG_CHECKPOINT_ONLINE
    1122            0 :                 || info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN
    1123              :             {
    1124            0 :                 let mut checkpoint_bytes = [0u8; pgv::xlog_utils::SIZEOF_CHECKPOINT];
    1125            0 :                 buf.copy_to_slice(&mut checkpoint_bytes);
    1126            0 :                 let xlog_checkpoint = pgv::CheckPoint::decode(&checkpoint_bytes)?;
    1127            0 :                 trace!(
    1128            0 :                     "xlog_checkpoint.oldestXid={}, checkpoint.oldestXid={}",
    1129              :                     xlog_checkpoint.oldestXid, cp.oldestXid
    1130              :                 );
    1131            0 :                 if (cp.oldestXid.wrapping_sub(xlog_checkpoint.oldestXid) as i32) < 0 {
    1132            0 :                     cp.oldestXid = xlog_checkpoint.oldestXid;
    1133            0 :                 }
    1134            0 :                 trace!(
    1135            0 :                     "xlog_checkpoint.oldestActiveXid={}, checkpoint.oldestActiveXid={}",
    1136              :                     xlog_checkpoint.oldestActiveXid, cp.oldestActiveXid
    1137              :                 );
    1138              : 
    1139              :                 // A shutdown checkpoint has `oldestActiveXid == InvalidTransactionid`,
    1140              :                 // because at shutdown, all in-progress transactions will implicitly
    1141              :                 // end. Postgres startup code knows that, and allows hot standby to start
    1142              :                 // immediately from a shutdown checkpoint.
    1143              :                 //
    1144              :                 // In Neon, Postgres hot standby startup always behaves as if starting from
    1145              :                 // an online checkpoint. It needs a valid `oldestActiveXid` value, so
    1146              :                 // instead of overwriting self.checkpoint.oldestActiveXid with
    1147              :                 // InvalidTransactionid from the checkpoint WAL record, update it to a
    1148              :                 // proper value, knowing that there are no in-progress transactions at this
    1149              :                 // point, except for prepared transactions.
    1150              :                 //
    1151              :                 // See also the neon code changes in the InitWalRecovery() function.
    1152            0 :                 if xlog_checkpoint.oldestActiveXid == pg_constants::INVALID_TRANSACTION_ID
    1153            0 :                     && info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN
    1154              :                 {
    1155            0 :                     let oldest_active_xid = if pg_version >= 17 {
    1156            0 :                         let mut oldest_active_full_xid = cp.nextXid.value;
    1157            0 :                         for xid in modification.tline.list_twophase_files(lsn, ctx).await? {
    1158            0 :                             if xid < oldest_active_full_xid {
    1159            0 :                                 oldest_active_full_xid = xid;
    1160            0 :                             }
    1161              :                         }
    1162            0 :                         oldest_active_full_xid as u32
    1163              :                     } else {
    1164            0 :                         let mut oldest_active_xid = cp.nextXid.value as u32;
    1165            0 :                         for xid in modification.tline.list_twophase_files(lsn, ctx).await? {
    1166            0 :                             let narrow_xid = xid as u32;
    1167            0 :                             if (narrow_xid.wrapping_sub(oldest_active_xid) as i32) < 0 {
    1168            0 :                                 oldest_active_xid = narrow_xid;
    1169            0 :                             }
    1170              :                         }
    1171            0 :                         oldest_active_xid
    1172              :                     };
    1173            0 :                     cp.oldestActiveXid = oldest_active_xid;
    1174            0 :                 } else {
    1175            0 :                     cp.oldestActiveXid = xlog_checkpoint.oldestActiveXid;
    1176            0 :                 }
    1177              :                 // NB: We abuse the Checkpoint.redo field:
    1178              :                 //
    1179              :                 // - In PostgreSQL, the Checkpoint struct doesn't store the information
    1180              :                 //   of whether this is an online checkpoint or a shutdown checkpoint. It's
    1181              :                 //   stored in the XLOG info field of the WAL record, shutdown checkpoints
    1182              :                 //   use record type XLOG_CHECKPOINT_SHUTDOWN and online checkpoints use
    1183              :                 //   XLOG_CHECKPOINT_ONLINE. We don't store the original WAL record headers
    1184              :                 //   in the pageserver, however.
    1185              :                 //
    1186              :                 // - In PostgreSQL, the Checkpoint.redo field stores the *start* of the
    1187              :                 //   checkpoint record, if it's a shutdown checkpoint. But when we are
    1188              :                 //   starting from a shutdown checkpoint, the basebackup LSN is the *end*
    1189              :                 //   of the shutdown checkpoint WAL record. That makes it difficult to
    1190              :                 //   correctly detect whether we're starting from a shutdown record or
    1191              :                 //   not.
    1192              :                 //
    1193              :                 // To address both of those issues, we store 0 in the redo field if it's
    1194              :                 // an online checkpoint record, and the record's *end* LSN if it's a
    1195              :                 // shutdown checkpoint. We don't need the original redo pointer in neon,
    1196              :                 // because we don't perform WAL replay at startup anyway, so we can get
    1197              :                 // away with abusing the redo field like this.
    1198              :                 //
    1199              :                 // XXX: Ideally, we would persist the extra information in a more
    1200              :                 // explicit format, rather than repurpose the fields of the Postgres
    1201              :                 // struct like this. However, we already have persisted data like this,
    1202              :                 // so we need to maintain backwards compatibility.
    1203              :                 //
    1204              :                 // NB: We didn't originally have this convention, so there are still old
    1205              :                 // persisted records that didn't do this. Before, we didn't update the
    1206              :                 // persisted redo field at all. That means that old records have a bogus
    1207              :                 // redo pointer that points to some old value, from the checkpoint record
    1208              :                 // that was originally imported from the data directory. If it was a
    1209              :                 // project created in Neon, that means it points to the first checkpoint
    1210              :                 // after initdb. That's OK for our purposes: all such old checkpoints are
    1211              :                 // treated as old online checkpoints when the basebackup is created.
    1212            0 :                 cp.redo = if info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN {
    1213              :                     // Store the *end* LSN of the checkpoint record. Or to be precise,
    1214              :                     // the start LSN of the *next* record, i.e. if the record ends
    1215              :                     // exactly at page boundary, the redo LSN points to just after the
    1216              :                     // page header on the next page.
    1217            0 :                     lsn.into()
    1218              :                 } else {
    1219            0 :                     Lsn::INVALID.into()
    1220              :                 };
    1221              : 
    1222              :                 // Write a new checkpoint key-value pair on every checkpoint record, even
    1223              :                 // if nothing really changed. Not strictly required, but it seems nice to
    1224              :                 // have some trace of the checkpoint records in the layer files at the same
    1225              :                 // LSNs.
    1226            0 :                 self.checkpoint_modified = true;
    1227            0 :             }
    1228              :         });
    1229              : 
    1230           60 :         Ok(())
    1231           60 :     }
    1232              : 
    1233            0 :     async fn ingest_logical_message_put(
    1234            0 :         &mut self,
    1235            0 :         put: PutLogicalMessage,
    1236            0 :         modification: &mut DatadirModification<'_>,
    1237            0 :         ctx: &RequestContext,
    1238            0 :     ) -> Result<()> {
    1239            0 :         let PutLogicalMessage { path, buf } = put;
    1240            0 :         modification.put_file(path.as_str(), &buf, ctx).await
    1241            0 :     }
    1242              : 
    1243            0 :     fn ingest_standby_record(&mut self, record: StandbyRecord) -> Result<()> {
    1244            0 :         match record {
    1245            0 :             StandbyRecord::RunningXacts(running_xacts) => {
    1246            0 :                 enum_pgversion_dispatch!(&mut self.checkpoint, CheckPoint, cp, {
    1247            0 :                     cp.oldestActiveXid = running_xacts.oldest_running_xid;
    1248            0 :                 });
    1249              : 
    1250            0 :                 self.checkpoint_modified = true;
    1251            0 :             }
    1252            0 :         }
    1253            0 : 
    1254            0 :         Ok(())
    1255            0 :     }
    1256              : 
    1257            0 :     async fn ingest_replorigin_record(
    1258            0 :         &mut self,
    1259            0 :         record: ReploriginRecord,
    1260            0 :         modification: &mut DatadirModification<'_>,
    1261            0 :     ) -> Result<()> {
    1262            0 :         match record {
    1263            0 :             ReploriginRecord::Set(set) => {
    1264            0 :                 modification
    1265            0 :                     .set_replorigin(set.node_id, set.remote_lsn)
    1266            0 :                     .await?;
    1267              :             }
    1268            0 :             ReploriginRecord::Drop(drop) => {
    1269            0 :                 modification.drop_replorigin(drop.node_id).await?;
    1270              :             }
    1271              :         }
    1272              : 
    1273            0 :         Ok(())
    1274            0 :     }
    1275              : 
    1276           36 :     async fn put_rel_creation(
    1277           36 :         &mut self,
    1278           36 :         modification: &mut DatadirModification<'_>,
    1279           36 :         rel: RelTag,
    1280           36 :         ctx: &RequestContext,
    1281           36 :     ) -> Result<()> {
    1282           36 :         modification.put_rel_creation(rel, 0, ctx).await?;
    1283           36 :         Ok(())
    1284           36 :     }
    1285              : 
    1286              :     #[cfg(test)]
    1287       544804 :     async fn put_rel_page_image(
    1288       544804 :         &mut self,
    1289       544804 :         modification: &mut DatadirModification<'_>,
    1290       544804 :         rel: RelTag,
    1291       544804 :         blknum: BlockNumber,
    1292       544804 :         img: Bytes,
    1293       544804 :         ctx: &RequestContext,
    1294       544804 :     ) -> Result<(), PageReconstructError> {
    1295       544804 :         self.handle_rel_extend(modification, rel, blknum, ctx)
    1296       544804 :             .await?;
    1297       544804 :         modification.put_rel_page_image(rel, blknum, img)?;
    1298       544804 :         Ok(())
    1299       544804 :     }
    1300              : 
    1301           24 :     async fn put_rel_wal_record(
    1302           24 :         &mut self,
    1303           24 :         modification: &mut DatadirModification<'_>,
    1304           24 :         rel: RelTag,
    1305           24 :         blknum: BlockNumber,
    1306           24 :         rec: NeonWalRecord,
    1307           24 :         ctx: &RequestContext,
    1308           24 :     ) -> Result<()> {
    1309           24 :         self.handle_rel_extend(modification, rel, blknum, ctx)
    1310           24 :             .await?;
    1311           24 :         modification.put_rel_wal_record(rel, blknum, rec)?;
    1312           24 :         Ok(())
    1313           24 :     }
    1314              : 
    1315        12024 :     async fn put_rel_truncation(
    1316        12024 :         &mut self,
    1317        12024 :         modification: &mut DatadirModification<'_>,
    1318        12024 :         rel: RelTag,
    1319        12024 :         nblocks: BlockNumber,
    1320        12024 :         ctx: &RequestContext,
    1321        12024 :     ) -> anyhow::Result<()> {
    1322        12024 :         modification.put_rel_truncation(rel, nblocks, ctx).await?;
    1323        12024 :         Ok(())
    1324        12024 :     }
    1325              : 
    1326       544828 :     async fn handle_rel_extend(
    1327       544828 :         &mut self,
    1328       544828 :         modification: &mut DatadirModification<'_>,
    1329       544828 :         rel: RelTag,
    1330       544828 :         blknum: BlockNumber,
    1331       544828 :         ctx: &RequestContext,
    1332       544828 :     ) -> Result<(), PageReconstructError> {
    1333       544828 :         let new_nblocks = blknum + 1;
    1334              :         // Check if the relation exists. We implicitly create relations on first
    1335              :         // record.
    1336       544828 :         let old_nblocks = modification.create_relation_if_required(rel, ctx).await?;
    1337              : 
    1338       544828 :         if new_nblocks > old_nblocks {
    1339              :             //info!("extending {} {} to {}", rel, old_nblocks, new_nblocks);
    1340       544796 :             modification.put_rel_extend(rel, new_nblocks, ctx).await?;
    1341              : 
    1342       544796 :             let mut key = rel_block_to_key(rel, blknum);
    1343       544796 : 
    1344       544796 :             // fill the gap with zeros
    1345       544796 :             let mut gap_blocks_filled: u64 = 0;
    1346       544796 :             for gap_blknum in old_nblocks..blknum {
    1347         5996 :                 key.field6 = gap_blknum;
    1348         5996 : 
    1349         5996 :                 if self.shard.get_shard_number(&key) != self.shard.number {
    1350            0 :                     continue;
    1351         5996 :                 }
    1352         5996 : 
    1353         5996 :                 modification.put_rel_page_image_zero(rel, gap_blknum)?;
    1354         5996 :                 gap_blocks_filled += 1;
    1355              :             }
    1356              : 
    1357       544796 :             WAL_INGEST
    1358       544796 :                 .gap_blocks_zeroed_on_rel_extend
    1359       544796 :                 .inc_by(gap_blocks_filled);
    1360       544796 : 
    1361       544796 :             // Log something when relation extends cause use to fill gaps
    1362       544796 :             // with zero pages. Logging is rate limited per pg version to
    1363       544796 :             // avoid skewing.
    1364       544796 :             if gap_blocks_filled > 0 {
    1365              :                 use std::sync::Mutex;
    1366              : 
    1367              :                 use once_cell::sync::Lazy;
    1368              :                 use utils::rate_limit::RateLimit;
    1369              : 
    1370              :                 struct RateLimitPerPgVersion {
    1371              :                     rate_limiters: [Lazy<Mutex<RateLimit>>; 4],
    1372              :                 }
    1373              : 
    1374              :                 impl RateLimitPerPgVersion {
    1375            0 :                     const fn new() -> Self {
    1376            0 :                         Self {
    1377            0 :                             rate_limiters: [const {
    1378            4 :                                 Lazy::new(|| Mutex::new(RateLimit::new(Duration::from_secs(30))))
    1379            0 :                             }; 4],
    1380            0 :                         }
    1381            0 :                     }
    1382              : 
    1383            8 :                     const fn rate_limiter(
    1384            8 :                         &self,
    1385            8 :                         pg_version: u32,
    1386            8 :                     ) -> Option<&Lazy<Mutex<RateLimit>>> {
    1387              :                         const MIN_PG_VERSION: u32 = 14;
    1388              :                         const MAX_PG_VERSION: u32 = 17;
    1389              : 
    1390            8 :                         if pg_version < MIN_PG_VERSION || pg_version > MAX_PG_VERSION {
    1391            0 :                             return None;
    1392            8 :                         }
    1393            8 : 
    1394            8 :                         Some(&self.rate_limiters[(pg_version - MIN_PG_VERSION) as usize])
    1395            8 :                     }
    1396              :                 }
    1397              : 
    1398              :                 static LOGGED: RateLimitPerPgVersion = RateLimitPerPgVersion::new();
    1399            8 :                 if let Some(rate_limiter) = LOGGED.rate_limiter(modification.tline.pg_version) {
    1400            8 :                     if let Ok(mut locked) = rate_limiter.try_lock() {
    1401            8 :                         locked.call(|| {
    1402            4 :                             info!(
    1403            0 :                                 lsn=%modification.get_lsn(),
    1404            0 :                                 pg_version=%modification.tline.pg_version,
    1405            0 :                                 rel=%rel,
    1406            0 :                                 "Filled {} gap blocks on rel extend to {} from {}",
    1407              :                                 gap_blocks_filled,
    1408              :                                 new_nblocks,
    1409              :                                 old_nblocks);
    1410            8 :                         });
    1411            8 :                     }
    1412            0 :                 }
    1413       544788 :             }
    1414           32 :         }
    1415       544828 :         Ok(())
    1416       544828 :     }
    1417              : 
    1418            0 :     async fn put_slru_page_image(
    1419            0 :         &mut self,
    1420            0 :         modification: &mut DatadirModification<'_>,
    1421            0 :         kind: SlruKind,
    1422            0 :         segno: u32,
    1423            0 :         blknum: BlockNumber,
    1424            0 :         img: Bytes,
    1425            0 :         ctx: &RequestContext,
    1426            0 :     ) -> Result<()> {
    1427            0 :         if !self.shard.is_shard_zero() {
    1428            0 :             return Ok(());
    1429            0 :         }
    1430            0 : 
    1431            0 :         self.handle_slru_extend(modification, kind, segno, blknum, ctx)
    1432            0 :             .await?;
    1433            0 :         modification.put_slru_page_image(kind, segno, blknum, img)?;
    1434            0 :         Ok(())
    1435            0 :     }
    1436              : 
    1437            0 :     async fn handle_slru_extend(
    1438            0 :         &mut self,
    1439            0 :         modification: &mut DatadirModification<'_>,
    1440            0 :         kind: SlruKind,
    1441            0 :         segno: u32,
    1442            0 :         blknum: BlockNumber,
    1443            0 :         ctx: &RequestContext,
    1444            0 :     ) -> anyhow::Result<()> {
    1445            0 :         // we don't use a cache for this like we do for relations. SLRUS are explcitly
    1446            0 :         // extended with ZEROPAGE records, not with commit records, so it happens
    1447            0 :         // a lot less frequently.
    1448            0 : 
    1449            0 :         let new_nblocks = blknum + 1;
    1450              :         // Check if the relation exists. We implicitly create relations on first
    1451              :         // record.
    1452              :         // TODO: would be nice if to be more explicit about it
    1453            0 :         let old_nblocks = if !modification
    1454            0 :             .tline
    1455            0 :             .get_slru_segment_exists(kind, segno, Version::Modified(modification), ctx)
    1456            0 :             .await?
    1457              :         {
    1458              :             // create it with 0 size initially, the logic below will extend it
    1459            0 :             modification
    1460            0 :                 .put_slru_segment_creation(kind, segno, 0, ctx)
    1461            0 :                 .await?;
    1462            0 :             0
    1463              :         } else {
    1464            0 :             modification
    1465            0 :                 .tline
    1466            0 :                 .get_slru_segment_size(kind, segno, Version::Modified(modification), ctx)
    1467            0 :                 .await?
    1468              :         };
    1469              : 
    1470            0 :         if new_nblocks > old_nblocks {
    1471            0 :             trace!(
    1472            0 :                 "extending SLRU {:?} seg {} from {} to {} blocks",
    1473              :                 kind, segno, old_nblocks, new_nblocks
    1474              :             );
    1475            0 :             modification.put_slru_extend(kind, segno, new_nblocks)?;
    1476              : 
    1477              :             // fill the gap with zeros
    1478            0 :             for gap_blknum in old_nblocks..blknum {
    1479            0 :                 modification.put_slru_page_image_zero(kind, segno, gap_blknum)?;
    1480              :             }
    1481            0 :         }
    1482            0 :         Ok(())
    1483            0 :     }
    1484              : }
    1485              : 
    1486              : /// Returns the size of the relation as of this modification, or None if the relation doesn't exist.
    1487              : ///
    1488              : /// This is only accurate on shard 0. On other shards, it will return the size up to the highest
    1489              : /// page number stored in the shard, or None if the shard does not have any pages for it.
    1490           24 : async fn get_relsize(
    1491           24 :     modification: &DatadirModification<'_>,
    1492           24 :     rel: RelTag,
    1493           24 :     ctx: &RequestContext,
    1494           24 : ) -> Result<Option<BlockNumber>, PageReconstructError> {
    1495           24 :     if !modification
    1496           24 :         .tline
    1497           24 :         .get_rel_exists(rel, Version::Modified(modification), ctx)
    1498           24 :         .await?
    1499              :     {
    1500            0 :         return Ok(None);
    1501           24 :     }
    1502           24 :     modification
    1503           24 :         .tline
    1504           24 :         .get_rel_size(rel, Version::Modified(modification), ctx)
    1505           24 :         .await
    1506           24 :         .map(Some)
    1507           24 : }
    1508              : 
    1509              : #[allow(clippy::bool_assert_comparison)]
    1510              : #[cfg(test)]
    1511              : mod tests {
    1512              :     use postgres_ffi::RELSEG_SIZE;
    1513              : 
    1514              :     use super::*;
    1515              :     use crate::DEFAULT_PG_VERSION;
    1516              :     use crate::tenant::harness::*;
    1517              :     use crate::tenant::remote_timeline_client::{INITDB_PATH, remote_initdb_archive_path};
    1518              :     use crate::tenant::storage_layer::IoConcurrency;
    1519              : 
    1520              :     /// Arbitrary relation tag, for testing.
    1521              :     const TESTREL_A: RelTag = RelTag {
    1522              :         spcnode: 0,
    1523              :         dbnode: 111,
    1524              :         relnode: 1000,
    1525              :         forknum: 0,
    1526              :     };
    1527              : 
    1528           24 :     fn assert_current_logical_size(_timeline: &Timeline, _lsn: Lsn) {
    1529           24 :         // TODO
    1530           24 :     }
    1531              : 
    1532              :     #[tokio::test]
    1533            4 :     async fn test_zeroed_checkpoint_decodes_correctly() -> Result<()> {
    1534           16 :         for i in 14..=16 {
    1535           12 :             dispatch_pgversion!(i, {
    1536            4 :                 pgv::CheckPoint::decode(&pgv::ZERO_CHECKPOINT)?;
    1537            4 :             });
    1538            4 :         }
    1539            4 : 
    1540            4 :         Ok(())
    1541            4 :     }
    1542              : 
    1543           16 :     async fn init_walingest_test(tline: &Timeline, ctx: &RequestContext) -> Result<WalIngest> {
    1544           16 :         let mut m = tline.begin_modification(Lsn(0x10));
    1545           16 :         m.put_checkpoint(dispatch_pgversion!(
    1546           16 :             tline.pg_version,
    1547            0 :             pgv::ZERO_CHECKPOINT.clone()
    1548            0 :         ))?;
    1549           16 :         m.put_relmap_file(0, 111, Bytes::from(""), ctx).await?; // dummy relmapper file
    1550           16 :         m.commit(ctx).await?;
    1551           16 :         let walingest = WalIngest::new(tline, Lsn(0x10), ctx).await?;
    1552              : 
    1553           16 :         Ok(walingest)
    1554           16 :     }
    1555              : 
    1556              :     #[tokio::test]
    1557            4 :     async fn test_relsize() -> Result<()> {
    1558            4 :         let (tenant, ctx) = TenantHarness::create("test_relsize").await?.load().await;
    1559            4 :         let io_concurrency = IoConcurrency::spawn_for_test();
    1560            4 :         let tline = tenant
    1561            4 :             .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)
    1562            4 :             .await?;
    1563            4 :         let mut walingest = init_walingest_test(&tline, &ctx).await?;
    1564            4 : 
    1565            4 :         let mut m = tline.begin_modification(Lsn(0x20));
    1566            4 :         walingest.put_rel_creation(&mut m, TESTREL_A, &ctx).await?;
    1567            4 :         walingest
    1568            4 :             .put_rel_page_image(&mut m, TESTREL_A, 0, test_img("foo blk 0 at 2"), &ctx)
    1569            4 :             .await?;
    1570            4 :         m.commit(&ctx).await?;
    1571            4 :         let mut m = tline.begin_modification(Lsn(0x30));
    1572            4 :         walingest
    1573            4 :             .put_rel_page_image(&mut m, TESTREL_A, 0, test_img("foo blk 0 at 3"), &ctx)
    1574            4 :             .await?;
    1575            4 :         m.commit(&ctx).await?;
    1576            4 :         let mut m = tline.begin_modification(Lsn(0x40));
    1577            4 :         walingest
    1578            4 :             .put_rel_page_image(&mut m, TESTREL_A, 1, test_img("foo blk 1 at 4"), &ctx)
    1579            4 :             .await?;
    1580            4 :         m.commit(&ctx).await?;
    1581            4 :         let mut m = tline.begin_modification(Lsn(0x50));
    1582            4 :         walingest
    1583            4 :             .put_rel_page_image(&mut m, TESTREL_A, 2, test_img("foo blk 2 at 5"), &ctx)
    1584            4 :             .await?;
    1585            4 :         m.commit(&ctx).await?;
    1586            4 : 
    1587            4 :         assert_current_logical_size(&tline, Lsn(0x50));
    1588            4 : 
    1589            4 :         let test_span = tracing::info_span!(parent: None, "test",
    1590            4 :                                             tenant_id=%tline.tenant_shard_id.tenant_id,
    1591            0 :                                             shard_id=%tline.tenant_shard_id.shard_slug(),
    1592            0 :                                             timeline_id=%tline.timeline_id);
    1593            4 : 
    1594            4 :         // The relation was created at LSN 2, not visible at LSN 1 yet.
    1595            4 :         assert_eq!(
    1596            4 :             tline
    1597            4 :                 .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
    1598            4 :                 .await?,
    1599            4 :             false
    1600            4 :         );
    1601            4 :         assert!(
    1602            4 :             tline
    1603            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
    1604            4 :                 .await
    1605            4 :                 .is_err()
    1606            4 :         );
    1607            4 :         assert_eq!(
    1608            4 :             tline
    1609            4 :                 .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
    1610            4 :                 .await?,
    1611            4 :             true
    1612            4 :         );
    1613            4 :         assert_eq!(
    1614            4 :             tline
    1615            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
    1616            4 :                 .await?,
    1617            4 :             1
    1618            4 :         );
    1619            4 :         assert_eq!(
    1620            4 :             tline
    1621            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x50)), &ctx)
    1622            4 :                 .await?,
    1623            4 :             3
    1624            4 :         );
    1625            4 : 
    1626            4 :         // Check page contents at each LSN
    1627            4 :         assert_eq!(
    1628            4 :             tline
    1629            4 :                 .get_rel_page_at_lsn(
    1630            4 :                     TESTREL_A,
    1631            4 :                     0,
    1632            4 :                     Version::Lsn(Lsn(0x20)),
    1633            4 :                     &ctx,
    1634            4 :                     io_concurrency.clone()
    1635            4 :                 )
    1636            4 :                 .instrument(test_span.clone())
    1637            4 :                 .await?,
    1638            4 :             test_img("foo blk 0 at 2")
    1639            4 :         );
    1640            4 : 
    1641            4 :         assert_eq!(
    1642            4 :             tline
    1643            4 :                 .get_rel_page_at_lsn(
    1644            4 :                     TESTREL_A,
    1645            4 :                     0,
    1646            4 :                     Version::Lsn(Lsn(0x30)),
    1647            4 :                     &ctx,
    1648            4 :                     io_concurrency.clone()
    1649            4 :                 )
    1650            4 :                 .instrument(test_span.clone())
    1651            4 :                 .await?,
    1652            4 :             test_img("foo blk 0 at 3")
    1653            4 :         );
    1654            4 : 
    1655            4 :         assert_eq!(
    1656            4 :             tline
    1657            4 :                 .get_rel_page_at_lsn(
    1658            4 :                     TESTREL_A,
    1659            4 :                     0,
    1660            4 :                     Version::Lsn(Lsn(0x40)),
    1661            4 :                     &ctx,
    1662            4 :                     io_concurrency.clone()
    1663            4 :                 )
    1664            4 :                 .instrument(test_span.clone())
    1665            4 :                 .await?,
    1666            4 :             test_img("foo blk 0 at 3")
    1667            4 :         );
    1668            4 :         assert_eq!(
    1669            4 :             tline
    1670            4 :                 .get_rel_page_at_lsn(
    1671            4 :                     TESTREL_A,
    1672            4 :                     1,
    1673            4 :                     Version::Lsn(Lsn(0x40)),
    1674            4 :                     &ctx,
    1675            4 :                     io_concurrency.clone()
    1676            4 :                 )
    1677            4 :                 .instrument(test_span.clone())
    1678            4 :                 .await?,
    1679            4 :             test_img("foo blk 1 at 4")
    1680            4 :         );
    1681            4 : 
    1682            4 :         assert_eq!(
    1683            4 :             tline
    1684            4 :                 .get_rel_page_at_lsn(
    1685            4 :                     TESTREL_A,
    1686            4 :                     0,
    1687            4 :                     Version::Lsn(Lsn(0x50)),
    1688            4 :                     &ctx,
    1689            4 :                     io_concurrency.clone()
    1690            4 :                 )
    1691            4 :                 .instrument(test_span.clone())
    1692            4 :                 .await?,
    1693            4 :             test_img("foo blk 0 at 3")
    1694            4 :         );
    1695            4 :         assert_eq!(
    1696            4 :             tline
    1697            4 :                 .get_rel_page_at_lsn(
    1698            4 :                     TESTREL_A,
    1699            4 :                     1,
    1700            4 :                     Version::Lsn(Lsn(0x50)),
    1701            4 :                     &ctx,
    1702            4 :                     io_concurrency.clone()
    1703            4 :                 )
    1704            4 :                 .instrument(test_span.clone())
    1705            4 :                 .await?,
    1706            4 :             test_img("foo blk 1 at 4")
    1707            4 :         );
    1708            4 :         assert_eq!(
    1709            4 :             tline
    1710            4 :                 .get_rel_page_at_lsn(
    1711            4 :                     TESTREL_A,
    1712            4 :                     2,
    1713            4 :                     Version::Lsn(Lsn(0x50)),
    1714            4 :                     &ctx,
    1715            4 :                     io_concurrency.clone()
    1716            4 :                 )
    1717            4 :                 .instrument(test_span.clone())
    1718            4 :                 .await?,
    1719            4 :             test_img("foo blk 2 at 5")
    1720            4 :         );
    1721            4 : 
    1722            4 :         // Truncate last block
    1723            4 :         let mut m = tline.begin_modification(Lsn(0x60));
    1724            4 :         walingest
    1725            4 :             .put_rel_truncation(&mut m, TESTREL_A, 2, &ctx)
    1726            4 :             .await?;
    1727            4 :         m.commit(&ctx).await?;
    1728            4 :         assert_current_logical_size(&tline, Lsn(0x60));
    1729            4 : 
    1730            4 :         // Check reported size and contents after truncation
    1731            4 :         assert_eq!(
    1732            4 :             tline
    1733            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x60)), &ctx)
    1734            4 :                 .await?,
    1735            4 :             2
    1736            4 :         );
    1737            4 :         assert_eq!(
    1738            4 :             tline
    1739            4 :                 .get_rel_page_at_lsn(
    1740            4 :                     TESTREL_A,
    1741            4 :                     0,
    1742            4 :                     Version::Lsn(Lsn(0x60)),
    1743            4 :                     &ctx,
    1744            4 :                     io_concurrency.clone()
    1745            4 :                 )
    1746            4 :                 .instrument(test_span.clone())
    1747            4 :                 .await?,
    1748            4 :             test_img("foo blk 0 at 3")
    1749            4 :         );
    1750            4 :         assert_eq!(
    1751            4 :             tline
    1752            4 :                 .get_rel_page_at_lsn(
    1753            4 :                     TESTREL_A,
    1754            4 :                     1,
    1755            4 :                     Version::Lsn(Lsn(0x60)),
    1756            4 :                     &ctx,
    1757            4 :                     io_concurrency.clone()
    1758            4 :                 )
    1759            4 :                 .instrument(test_span.clone())
    1760            4 :                 .await?,
    1761            4 :             test_img("foo blk 1 at 4")
    1762            4 :         );
    1763            4 : 
    1764            4 :         // should still see the truncated block with older LSN
    1765            4 :         assert_eq!(
    1766            4 :             tline
    1767            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x50)), &ctx)
    1768            4 :                 .await?,
    1769            4 :             3
    1770            4 :         );
    1771            4 :         assert_eq!(
    1772            4 :             tline
    1773            4 :                 .get_rel_page_at_lsn(
    1774            4 :                     TESTREL_A,
    1775            4 :                     2,
    1776            4 :                     Version::Lsn(Lsn(0x50)),
    1777            4 :                     &ctx,
    1778            4 :                     io_concurrency.clone()
    1779            4 :                 )
    1780            4 :                 .instrument(test_span.clone())
    1781            4 :                 .await?,
    1782            4 :             test_img("foo blk 2 at 5")
    1783            4 :         );
    1784            4 : 
    1785            4 :         // Truncate to zero length
    1786            4 :         let mut m = tline.begin_modification(Lsn(0x68));
    1787            4 :         walingest
    1788            4 :             .put_rel_truncation(&mut m, TESTREL_A, 0, &ctx)
    1789            4 :             .await?;
    1790            4 :         m.commit(&ctx).await?;
    1791            4 :         assert_eq!(
    1792            4 :             tline
    1793            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x68)), &ctx)
    1794            4 :                 .await?,
    1795            4 :             0
    1796            4 :         );
    1797            4 : 
    1798            4 :         // Extend from 0 to 2 blocks, leaving a gap
    1799            4 :         let mut m = tline.begin_modification(Lsn(0x70));
    1800            4 :         walingest
    1801            4 :             .put_rel_page_image(&mut m, TESTREL_A, 1, test_img("foo blk 1"), &ctx)
    1802            4 :             .await?;
    1803            4 :         m.commit(&ctx).await?;
    1804            4 :         assert_eq!(
    1805            4 :             tline
    1806            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x70)), &ctx)
    1807            4 :                 .await?,
    1808            4 :             2
    1809            4 :         );
    1810            4 :         assert_eq!(
    1811            4 :             tline
    1812            4 :                 .get_rel_page_at_lsn(
    1813            4 :                     TESTREL_A,
    1814            4 :                     0,
    1815            4 :                     Version::Lsn(Lsn(0x70)),
    1816            4 :                     &ctx,
    1817            4 :                     io_concurrency.clone()
    1818            4 :                 )
    1819            4 :                 .instrument(test_span.clone())
    1820            4 :                 .await?,
    1821            4 :             ZERO_PAGE
    1822            4 :         );
    1823            4 :         assert_eq!(
    1824            4 :             tline
    1825            4 :                 .get_rel_page_at_lsn(
    1826            4 :                     TESTREL_A,
    1827            4 :                     1,
    1828            4 :                     Version::Lsn(Lsn(0x70)),
    1829            4 :                     &ctx,
    1830            4 :                     io_concurrency.clone()
    1831            4 :                 )
    1832            4 :                 .instrument(test_span.clone())
    1833            4 :                 .await?,
    1834            4 :             test_img("foo blk 1")
    1835            4 :         );
    1836            4 : 
    1837            4 :         // Extend a lot more, leaving a big gap that spans across segments
    1838            4 :         let mut m = tline.begin_modification(Lsn(0x80));
    1839            4 :         walingest
    1840            4 :             .put_rel_page_image(&mut m, TESTREL_A, 1500, test_img("foo blk 1500"), &ctx)
    1841            4 :             .await?;
    1842            4 :         m.commit(&ctx).await?;
    1843            4 :         assert_eq!(
    1844            4 :             tline
    1845            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x80)), &ctx)
    1846            4 :                 .await?,
    1847            4 :             1501
    1848            4 :         );
    1849         5996 :         for blk in 2..1500 {
    1850         5992 :             assert_eq!(
    1851         5992 :                 tline
    1852         5992 :                     .get_rel_page_at_lsn(
    1853         5992 :                         TESTREL_A,
    1854         5992 :                         blk,
    1855         5992 :                         Version::Lsn(Lsn(0x80)),
    1856         5992 :                         &ctx,
    1857         5992 :                         io_concurrency.clone()
    1858         5992 :                     )
    1859         5992 :                     .instrument(test_span.clone())
    1860         5992 :                     .await?,
    1861         5992 :                 ZERO_PAGE
    1862            4 :             );
    1863            4 :         }
    1864            4 :         assert_eq!(
    1865            4 :             tline
    1866            4 :                 .get_rel_page_at_lsn(
    1867            4 :                     TESTREL_A,
    1868            4 :                     1500,
    1869            4 :                     Version::Lsn(Lsn(0x80)),
    1870            4 :                     &ctx,
    1871            4 :                     io_concurrency.clone()
    1872            4 :                 )
    1873            4 :                 .instrument(test_span.clone())
    1874            4 :                 .await?,
    1875            4 :             test_img("foo blk 1500")
    1876            4 :         );
    1877            4 : 
    1878            4 :         Ok(())
    1879            4 :     }
    1880              : 
    1881              :     // Test what happens if we dropped a relation
    1882              :     // and then created it again within the same layer.
    1883              :     #[tokio::test]
    1884            4 :     async fn test_drop_extend() -> Result<()> {
    1885            4 :         let (tenant, ctx) = TenantHarness::create("test_drop_extend")
    1886            4 :             .await?
    1887            4 :             .load()
    1888            4 :             .await;
    1889            4 :         let tline = tenant
    1890            4 :             .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)
    1891            4 :             .await?;
    1892            4 :         let mut walingest = init_walingest_test(&tline, &ctx).await?;
    1893            4 : 
    1894            4 :         let mut m = tline.begin_modification(Lsn(0x20));
    1895            4 :         walingest
    1896            4 :             .put_rel_page_image(&mut m, TESTREL_A, 0, test_img("foo blk 0 at 2"), &ctx)
    1897            4 :             .await?;
    1898            4 :         m.commit(&ctx).await?;
    1899            4 : 
    1900            4 :         // Check that rel exists and size is correct
    1901            4 :         assert_eq!(
    1902            4 :             tline
    1903            4 :                 .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
    1904            4 :                 .await?,
    1905            4 :             true
    1906            4 :         );
    1907            4 :         assert_eq!(
    1908            4 :             tline
    1909            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
    1910            4 :                 .await?,
    1911            4 :             1
    1912            4 :         );
    1913            4 : 
    1914            4 :         // Drop rel
    1915            4 :         let mut m = tline.begin_modification(Lsn(0x30));
    1916            4 :         let mut rel_drops = HashMap::new();
    1917            4 :         rel_drops.insert((TESTREL_A.spcnode, TESTREL_A.dbnode), vec![TESTREL_A]);
    1918            4 :         m.put_rel_drops(rel_drops, &ctx).await?;
    1919            4 :         m.commit(&ctx).await?;
    1920            4 : 
    1921            4 :         // Check that rel is not visible anymore
    1922            4 :         assert_eq!(
    1923            4 :             tline
    1924            4 :                 .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x30)), &ctx)
    1925            4 :                 .await?,
    1926            4 :             false
    1927            4 :         );
    1928            4 : 
    1929            4 :         // FIXME: should fail
    1930            4 :         //assert!(tline.get_rel_size(TESTREL_A, Lsn(0x30), false)?.is_none());
    1931            4 : 
    1932            4 :         // Re-create it
    1933            4 :         let mut m = tline.begin_modification(Lsn(0x40));
    1934            4 :         walingest
    1935            4 :             .put_rel_page_image(&mut m, TESTREL_A, 0, test_img("foo blk 0 at 4"), &ctx)
    1936            4 :             .await?;
    1937            4 :         m.commit(&ctx).await?;
    1938            4 : 
    1939            4 :         // Check that rel exists and size is correct
    1940            4 :         assert_eq!(
    1941            4 :             tline
    1942            4 :                 .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x40)), &ctx)
    1943            4 :                 .await?,
    1944            4 :             true
    1945            4 :         );
    1946            4 :         assert_eq!(
    1947            4 :             tline
    1948            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x40)), &ctx)
    1949            4 :                 .await?,
    1950            4 :             1
    1951            4 :         );
    1952            4 : 
    1953            4 :         Ok(())
    1954            4 :     }
    1955              : 
    1956              :     // Test what happens if we truncated a relation
    1957              :     // so that one of its segments was dropped
    1958              :     // and then extended it again within the same layer.
    1959              :     #[tokio::test]
    1960            4 :     async fn test_truncate_extend() -> Result<()> {
    1961            4 :         let (tenant, ctx) = TenantHarness::create("test_truncate_extend")
    1962            4 :             .await?
    1963            4 :             .load()
    1964            4 :             .await;
    1965            4 :         let io_concurrency = IoConcurrency::spawn_for_test();
    1966            4 :         let tline = tenant
    1967            4 :             .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)
    1968            4 :             .await?;
    1969            4 :         let mut walingest = init_walingest_test(&tline, &ctx).await?;
    1970            4 : 
    1971            4 :         // Create a 20 MB relation (the size is arbitrary)
    1972            4 :         let relsize = 20 * 1024 * 1024 / 8192;
    1973            4 :         let mut m = tline.begin_modification(Lsn(0x20));
    1974        10240 :         for blkno in 0..relsize {
    1975        10240 :             let data = format!("foo blk {} at {}", blkno, Lsn(0x20));
    1976        10240 :             walingest
    1977        10240 :                 .put_rel_page_image(&mut m, TESTREL_A, blkno, test_img(&data), &ctx)
    1978        10240 :                 .await?;
    1979            4 :         }
    1980            4 :         m.commit(&ctx).await?;
    1981            4 : 
    1982            4 :         let test_span = tracing::info_span!(parent: None, "test",
    1983            4 :                                             tenant_id=%tline.tenant_shard_id.tenant_id,
    1984            0 :                                             shard_id=%tline.tenant_shard_id.shard_slug(),
    1985            0 :                                             timeline_id=%tline.timeline_id);
    1986            4 : 
    1987            4 :         // The relation was created at LSN 20, not visible at LSN 1 yet.
    1988            4 :         assert_eq!(
    1989            4 :             tline
    1990            4 :                 .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
    1991            4 :                 .await?,
    1992            4 :             false
    1993            4 :         );
    1994            4 :         assert!(
    1995            4 :             tline
    1996            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x10)), &ctx)
    1997            4 :                 .await
    1998            4 :                 .is_err()
    1999            4 :         );
    2000            4 : 
    2001            4 :         assert_eq!(
    2002            4 :             tline
    2003            4 :                 .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
    2004            4 :                 .await?,
    2005            4 :             true
    2006            4 :         );
    2007            4 :         assert_eq!(
    2008            4 :             tline
    2009            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x20)), &ctx)
    2010            4 :                 .await?,
    2011            4 :             relsize
    2012            4 :         );
    2013            4 : 
    2014            4 :         // Check relation content
    2015        10240 :         for blkno in 0..relsize {
    2016        10240 :             let lsn = Lsn(0x20);
    2017        10240 :             let data = format!("foo blk {} at {}", blkno, lsn);
    2018        10240 :             assert_eq!(
    2019        10240 :                 tline
    2020        10240 :                     .get_rel_page_at_lsn(
    2021        10240 :                         TESTREL_A,
    2022        10240 :                         blkno,
    2023        10240 :                         Version::Lsn(lsn),
    2024        10240 :                         &ctx,
    2025        10240 :                         io_concurrency.clone()
    2026        10240 :                     )
    2027        10240 :                     .instrument(test_span.clone())
    2028        10240 :                     .await?,
    2029        10240 :                 test_img(&data)
    2030            4 :             );
    2031            4 :         }
    2032            4 : 
    2033            4 :         // Truncate relation so that second segment was dropped
    2034            4 :         // - only leave one page
    2035            4 :         let mut m = tline.begin_modification(Lsn(0x60));
    2036            4 :         walingest
    2037            4 :             .put_rel_truncation(&mut m, TESTREL_A, 1, &ctx)
    2038            4 :             .await?;
    2039            4 :         m.commit(&ctx).await?;
    2040            4 : 
    2041            4 :         // Check reported size and contents after truncation
    2042            4 :         assert_eq!(
    2043            4 :             tline
    2044            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x60)), &ctx)
    2045            4 :                 .await?,
    2046            4 :             1
    2047            4 :         );
    2048            4 : 
    2049            8 :         for blkno in 0..1 {
    2050            4 :             let lsn = Lsn(0x20);
    2051            4 :             let data = format!("foo blk {} at {}", blkno, lsn);
    2052            4 :             assert_eq!(
    2053            4 :                 tline
    2054            4 :                     .get_rel_page_at_lsn(
    2055            4 :                         TESTREL_A,
    2056            4 :                         blkno,
    2057            4 :                         Version::Lsn(Lsn(0x60)),
    2058            4 :                         &ctx,
    2059            4 :                         io_concurrency.clone()
    2060            4 :                     )
    2061            4 :                     .instrument(test_span.clone())
    2062            4 :                     .await?,
    2063            4 :                 test_img(&data)
    2064            4 :             );
    2065            4 :         }
    2066            4 : 
    2067            4 :         // should still see all blocks with older LSN
    2068            4 :         assert_eq!(
    2069            4 :             tline
    2070            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x50)), &ctx)
    2071            4 :                 .await?,
    2072            4 :             relsize
    2073            4 :         );
    2074        10240 :         for blkno in 0..relsize {
    2075        10240 :             let lsn = Lsn(0x20);
    2076        10240 :             let data = format!("foo blk {} at {}", blkno, lsn);
    2077        10240 :             assert_eq!(
    2078        10240 :                 tline
    2079        10240 :                     .get_rel_page_at_lsn(
    2080        10240 :                         TESTREL_A,
    2081        10240 :                         blkno,
    2082        10240 :                         Version::Lsn(Lsn(0x50)),
    2083        10240 :                         &ctx,
    2084        10240 :                         io_concurrency.clone()
    2085        10240 :                     )
    2086        10240 :                     .instrument(test_span.clone())
    2087        10240 :                     .await?,
    2088        10240 :                 test_img(&data)
    2089            4 :             );
    2090            4 :         }
    2091            4 : 
    2092            4 :         // Extend relation again.
    2093            4 :         // Add enough blocks to create second segment
    2094            4 :         let lsn = Lsn(0x80);
    2095            4 :         let mut m = tline.begin_modification(lsn);
    2096        10240 :         for blkno in 0..relsize {
    2097        10240 :             let data = format!("foo blk {} at {}", blkno, lsn);
    2098        10240 :             walingest
    2099        10240 :                 .put_rel_page_image(&mut m, TESTREL_A, blkno, test_img(&data), &ctx)
    2100        10240 :                 .await?;
    2101            4 :         }
    2102            4 :         m.commit(&ctx).await?;
    2103            4 : 
    2104            4 :         assert_eq!(
    2105            4 :             tline
    2106            4 :                 .get_rel_exists(TESTREL_A, Version::Lsn(Lsn(0x80)), &ctx)
    2107            4 :                 .await?,
    2108            4 :             true
    2109            4 :         );
    2110            4 :         assert_eq!(
    2111            4 :             tline
    2112            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(0x80)), &ctx)
    2113            4 :                 .await?,
    2114            4 :             relsize
    2115            4 :         );
    2116            4 :         // Check relation content
    2117        10240 :         for blkno in 0..relsize {
    2118        10240 :             let lsn = Lsn(0x80);
    2119        10240 :             let data = format!("foo blk {} at {}", blkno, lsn);
    2120        10240 :             assert_eq!(
    2121        10240 :                 tline
    2122        10240 :                     .get_rel_page_at_lsn(
    2123        10240 :                         TESTREL_A,
    2124        10240 :                         blkno,
    2125        10240 :                         Version::Lsn(Lsn(0x80)),
    2126        10240 :                         &ctx,
    2127        10240 :                         io_concurrency.clone()
    2128        10240 :                     )
    2129        10240 :                     .instrument(test_span.clone())
    2130        10240 :                     .await?,
    2131        10240 :                 test_img(&data)
    2132            4 :             );
    2133            4 :         }
    2134            4 : 
    2135            4 :         Ok(())
    2136            4 :     }
    2137              : 
    2138              :     /// Test get_relsize() and truncation with a file larger than 1 GB, so that it's
    2139              :     /// split into multiple 1 GB segments in Postgres.
    2140              :     #[tokio::test]
    2141            4 :     async fn test_large_rel() -> Result<()> {
    2142            4 :         let (tenant, ctx) = TenantHarness::create("test_large_rel").await?.load().await;
    2143            4 :         let tline = tenant
    2144            4 :             .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)
    2145            4 :             .await?;
    2146            4 :         let mut walingest = init_walingest_test(&tline, &ctx).await?;
    2147            4 : 
    2148            4 :         let mut lsn = 0x10;
    2149       524292 :         for blknum in 0..RELSEG_SIZE + 1 {
    2150       524292 :             lsn += 0x10;
    2151       524292 :             let mut m = tline.begin_modification(Lsn(lsn));
    2152       524292 :             let img = test_img(&format!("foo blk {} at {}", blknum, Lsn(lsn)));
    2153       524292 :             walingest
    2154       524292 :                 .put_rel_page_image(&mut m, TESTREL_A, blknum as BlockNumber, img, &ctx)
    2155       524292 :                 .await?;
    2156       524292 :             m.commit(&ctx).await?;
    2157            4 :         }
    2158            4 : 
    2159            4 :         assert_current_logical_size(&tline, Lsn(lsn));
    2160            4 : 
    2161            4 :         assert_eq!(
    2162            4 :             tline
    2163            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
    2164            4 :                 .await?,
    2165            4 :             RELSEG_SIZE + 1
    2166            4 :         );
    2167            4 : 
    2168            4 :         // Truncate one block
    2169            4 :         lsn += 0x10;
    2170            4 :         let mut m = tline.begin_modification(Lsn(lsn));
    2171            4 :         walingest
    2172            4 :             .put_rel_truncation(&mut m, TESTREL_A, RELSEG_SIZE, &ctx)
    2173            4 :             .await?;
    2174            4 :         m.commit(&ctx).await?;
    2175            4 :         assert_eq!(
    2176            4 :             tline
    2177            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
    2178            4 :                 .await?,
    2179            4 :             RELSEG_SIZE
    2180            4 :         );
    2181            4 :         assert_current_logical_size(&tline, Lsn(lsn));
    2182            4 : 
    2183            4 :         // Truncate another block
    2184            4 :         lsn += 0x10;
    2185            4 :         let mut m = tline.begin_modification(Lsn(lsn));
    2186            4 :         walingest
    2187            4 :             .put_rel_truncation(&mut m, TESTREL_A, RELSEG_SIZE - 1, &ctx)
    2188            4 :             .await?;
    2189            4 :         m.commit(&ctx).await?;
    2190            4 :         assert_eq!(
    2191            4 :             tline
    2192            4 :                 .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
    2193            4 :                 .await?,
    2194            4 :             RELSEG_SIZE - 1
    2195            4 :         );
    2196            4 :         assert_current_logical_size(&tline, Lsn(lsn));
    2197            4 : 
    2198            4 :         // Truncate to 1500, and then truncate all the way down to 0, one block at a time
    2199            4 :         // This tests the behavior at segment boundaries
    2200            4 :         let mut size: i32 = 3000;
    2201        12008 :         while size >= 0 {
    2202        12004 :             lsn += 0x10;
    2203        12004 :             let mut m = tline.begin_modification(Lsn(lsn));
    2204        12004 :             walingest
    2205        12004 :                 .put_rel_truncation(&mut m, TESTREL_A, size as BlockNumber, &ctx)
    2206        12004 :                 .await?;
    2207        12004 :             m.commit(&ctx).await?;
    2208        12004 :             assert_eq!(
    2209        12004 :                 tline
    2210        12004 :                     .get_rel_size(TESTREL_A, Version::Lsn(Lsn(lsn)), &ctx)
    2211        12004 :                     .await?,
    2212        12004 :                 size as BlockNumber
    2213            4 :             );
    2214            4 : 
    2215        12004 :             size -= 1;
    2216            4 :         }
    2217            4 :         assert_current_logical_size(&tline, Lsn(lsn));
    2218            4 : 
    2219            4 :         Ok(())
    2220            4 :     }
    2221              : 
    2222              :     /// Replay a wal segment file taken directly from safekeepers.
    2223              :     ///
    2224              :     /// This test is useful for benchmarking since it allows us to profile only
    2225              :     /// the walingest code in a single-threaded executor, and iterate more quickly
    2226              :     /// without waiting for unrelated steps.
    2227              :     #[tokio::test]
    2228            4 :     async fn test_ingest_real_wal() {
    2229            4 :         use postgres_ffi::WAL_SEGMENT_SIZE;
    2230            4 :         use postgres_ffi::waldecoder::WalStreamDecoder;
    2231            4 : 
    2232            4 :         use crate::tenant::harness::*;
    2233            4 : 
    2234            4 :         // Define test data path and constants.
    2235            4 :         //
    2236            4 :         // Steps to reconstruct the data, if needed:
    2237            4 :         // 1. Run the pgbench python test
    2238            4 :         // 2. Take the first wal segment file from safekeeper
    2239            4 :         // 3. Compress it using `zstd --long input_file`
    2240            4 :         // 4. Copy initdb.tar.zst from local_fs_remote_storage
    2241            4 :         // 5. Grep sk logs for "restart decoder" to get startpoint
    2242            4 :         // 6. Run just the decoder from this test to get the endpoint.
    2243            4 :         //    It's the last LSN the decoder will output.
    2244            4 :         let pg_version = 15; // The test data was generated by pg15
    2245            4 :         let path = "test_data/sk_wal_segment_from_pgbench";
    2246            4 :         let wal_segment_path = format!("{path}/000000010000000000000001.zst");
    2247            4 :         let source_initdb_path = format!("{path}/{INITDB_PATH}");
    2248            4 :         let startpoint = Lsn::from_hex("14AEC08").unwrap();
    2249            4 :         let _endpoint = Lsn::from_hex("1FFFF98").unwrap();
    2250            4 : 
    2251            4 :         let harness = TenantHarness::create("test_ingest_real_wal").await.unwrap();
    2252            4 :         let span = harness
    2253            4 :             .span()
    2254            4 :             .in_scope(|| info_span!("timeline_span", timeline_id=%TIMELINE_ID));
    2255            4 :         let (tenant, ctx) = harness.load().await;
    2256            4 : 
    2257            4 :         let remote_initdb_path =
    2258            4 :             remote_initdb_archive_path(&tenant.tenant_shard_id().tenant_id, &TIMELINE_ID);
    2259            4 :         let initdb_path = harness.remote_fs_dir.join(remote_initdb_path.get_path());
    2260            4 : 
    2261            4 :         std::fs::create_dir_all(initdb_path.parent().unwrap())
    2262            4 :             .expect("creating test dir should work");
    2263            4 :         std::fs::copy(source_initdb_path, initdb_path).expect("copying the initdb.tar.zst works");
    2264            4 : 
    2265            4 :         // Bootstrap a real timeline. We can't use create_test_timeline because
    2266            4 :         // it doesn't create a real checkpoint, and Walingest::new tries to parse
    2267            4 :         // the garbage data.
    2268            4 :         let tline = tenant
    2269            4 :             .bootstrap_timeline_test(TIMELINE_ID, pg_version, Some(TIMELINE_ID), &ctx)
    2270            4 :             .await
    2271            4 :             .unwrap();
    2272            4 : 
    2273            4 :         // We fully read and decompress this into memory before decoding
    2274            4 :         // to get a more accurate perf profile of the decoder.
    2275            4 :         let bytes = {
    2276            4 :             use async_compression::tokio::bufread::ZstdDecoder;
    2277            4 :             let file = tokio::fs::File::open(wal_segment_path).await.unwrap();
    2278            4 :             let reader = tokio::io::BufReader::new(file);
    2279            4 :             let decoder = ZstdDecoder::new(reader);
    2280            4 :             let mut reader = tokio::io::BufReader::new(decoder);
    2281            4 :             let mut buffer = Vec::new();
    2282            4 :             tokio::io::copy_buf(&mut reader, &mut buffer).await.unwrap();
    2283            4 :             buffer
    2284            4 :         };
    2285            4 : 
    2286            4 :         // TODO start a profiler too
    2287            4 :         let started_at = std::time::Instant::now();
    2288            4 : 
    2289            4 :         // Initialize walingest
    2290            4 :         let xlogoff: usize = startpoint.segment_offset(WAL_SEGMENT_SIZE);
    2291            4 :         let mut decoder = WalStreamDecoder::new(startpoint, pg_version);
    2292            4 :         let mut walingest = WalIngest::new(tline.as_ref(), startpoint, &ctx)
    2293            4 :             .await
    2294            4 :             .unwrap();
    2295            4 :         let mut modification = tline.begin_modification(startpoint);
    2296            4 :         println!("decoding {} bytes", bytes.len() - xlogoff);
    2297            4 : 
    2298            4 :         // Decode and ingest wal. We process the wal in chunks because
    2299            4 :         // that's what happens when we get bytes from safekeepers.
    2300       949372 :         for chunk in bytes[xlogoff..].chunks(50) {
    2301       949372 :             decoder.feed_bytes(chunk);
    2302      1241072 :             while let Some((lsn, recdata)) = decoder.poll_decode().unwrap() {
    2303       291700 :                 let interpreted = InterpretedWalRecord::from_bytes_filtered(
    2304       291700 :                     recdata,
    2305       291700 :                     &[*modification.tline.get_shard_identity()],
    2306       291700 :                     lsn,
    2307       291700 :                     modification.tline.pg_version,
    2308       291700 :                 )
    2309       291700 :                 .unwrap()
    2310       291700 :                 .remove(modification.tline.get_shard_identity())
    2311       291700 :                 .unwrap();
    2312       291700 : 
    2313       291700 :                 walingest
    2314       291700 :                     .ingest_record(interpreted, &mut modification, &ctx)
    2315       291700 :                     .instrument(span.clone())
    2316       291700 :                     .await
    2317       291700 :                     .unwrap();
    2318            4 :             }
    2319       949372 :             modification.commit(&ctx).await.unwrap();
    2320            4 :         }
    2321            4 : 
    2322            4 :         let duration = started_at.elapsed();
    2323            4 :         println!("done in {:?}", duration);
    2324            4 :     }
    2325              : }
        

Generated by: LCOV version 2.1-beta