LCOV - code coverage report
Current view: top level - pageserver/src/tenant/remote_timeline_client - download.rs (source / functions) Coverage Total Hit
Test: 1b0a6a0c05cee5a7de360813c8034804e105ce1c.info Lines: 87.6 % 459 402
Test Date: 2025-03-12 00:01:28 Functions: 49.6 % 113 56

            Line data    Source code
       1              : //! Helper functions to download files from remote storage with a RemoteStorage
       2              : //!
       3              : //! The functions in this module retry failed operations automatically, according
       4              : //! to the FAILED_DOWNLOAD_RETRIES constant.
       5              : 
       6              : use std::collections::HashSet;
       7              : use std::future::Future;
       8              : use std::str::FromStr;
       9              : use std::time::SystemTime;
      10              : 
      11              : use anyhow::{Context, anyhow};
      12              : use camino::{Utf8Path, Utf8PathBuf};
      13              : use pageserver_api::shard::TenantShardId;
      14              : use remote_storage::{
      15              :     DownloadError, DownloadKind, DownloadOpts, GenericRemoteStorage, ListingMode, RemotePath,
      16              : };
      17              : use tokio::fs::{self, File, OpenOptions};
      18              : use tokio::io::{AsyncSeekExt, AsyncWriteExt};
      19              : use tokio_util::io::StreamReader;
      20              : use tokio_util::sync::CancellationToken;
      21              : use tracing::{info_span, warn};
      22              : use utils::crashsafe::path_with_suffix_extension;
      23              : use utils::id::{TenantId, TimelineId};
      24              : use utils::{backoff, pausable_failpoint};
      25              : 
      26              : use super::index::{IndexPart, LayerFileMetadata};
      27              : use super::manifest::TenantManifest;
      28              : use super::{
      29              :     FAILED_DOWNLOAD_WARN_THRESHOLD, FAILED_REMOTE_OP_RETRIES, INITDB_PATH, parse_remote_index_path,
      30              :     parse_remote_tenant_manifest_path, remote_index_path, remote_initdb_archive_path,
      31              :     remote_initdb_preserved_archive_path, remote_tenant_manifest_path,
      32              :     remote_tenant_manifest_prefix, remote_tenant_path,
      33              : };
      34              : use crate::TEMP_FILE_SUFFIX;
      35              : use crate::config::PageServerConf;
      36              : use crate::context::RequestContext;
      37              : use crate::span::{
      38              :     debug_assert_current_span_has_tenant_and_timeline_id, debug_assert_current_span_has_tenant_id,
      39              : };
      40              : use crate::tenant::Generation;
      41              : use crate::tenant::remote_timeline_client::{remote_layer_path, remote_timelines_path};
      42              : use crate::tenant::storage_layer::LayerName;
      43              : use crate::virtual_file::{MaybeFatalIo, VirtualFile, on_fatal_io_error};
      44              : 
      45              : ///
      46              : /// If 'metadata' is given, we will validate that the downloaded file's size matches that
      47              : /// in the metadata. (In the future, we might do more cross-checks, like CRC validation)
      48              : ///
      49              : /// Returns the size of the downloaded file.
      50              : #[allow(clippy::too_many_arguments)]
      51           28 : pub async fn download_layer_file<'a>(
      52           28 :     conf: &'static PageServerConf,
      53           28 :     storage: &'a GenericRemoteStorage,
      54           28 :     tenant_shard_id: TenantShardId,
      55           28 :     timeline_id: TimelineId,
      56           28 :     layer_file_name: &'a LayerName,
      57           28 :     layer_metadata: &'a LayerFileMetadata,
      58           28 :     local_path: &Utf8Path,
      59           28 :     gate: &utils::sync::gate::Gate,
      60           28 :     cancel: &CancellationToken,
      61           28 :     ctx: &RequestContext,
      62           28 : ) -> Result<u64, DownloadError> {
      63           28 :     debug_assert_current_span_has_tenant_and_timeline_id();
      64           28 : 
      65           28 :     let timeline_path = conf.timeline_path(&tenant_shard_id, &timeline_id);
      66           28 : 
      67           28 :     let remote_path = remote_layer_path(
      68           28 :         &tenant_shard_id.tenant_id,
      69           28 :         &timeline_id,
      70           28 :         layer_metadata.shard,
      71           28 :         layer_file_name,
      72           28 :         layer_metadata.generation,
      73           28 :     );
      74           28 : 
      75           28 :     // Perform a rename inspired by durable_rename from file_utils.c.
      76           28 :     // The sequence:
      77           28 :     //     write(tmp)
      78           28 :     //     fsync(tmp)
      79           28 :     //     rename(tmp, new)
      80           28 :     //     fsync(new)
      81           28 :     //     fsync(parent)
      82           28 :     // For more context about durable_rename check this email from postgres mailing list:
      83           28 :     // https://www.postgresql.org/message-id/56583BDD.9060302@2ndquadrant.com
      84           28 :     // If pageserver crashes the temp file will be deleted on startup and re-downloaded.
      85           28 :     let temp_file_path = path_with_suffix_extension(local_path, TEMP_DOWNLOAD_EXTENSION);
      86              : 
      87           28 :     let bytes_amount = download_retry(
      88           28 :         || async {
      89           28 :             download_object(storage, &remote_path, &temp_file_path, gate, cancel, ctx).await
      90           56 :         },
      91           28 :         &format!("download {remote_path:?}"),
      92           28 :         cancel,
      93           28 :     )
      94           28 :     .await?;
      95              : 
      96           28 :     let expected = layer_metadata.file_size;
      97           28 :     if expected != bytes_amount {
      98            0 :         return Err(DownloadError::Other(anyhow!(
      99            0 :             "According to layer file metadata should have downloaded {expected} bytes but downloaded {bytes_amount} bytes into file {temp_file_path:?}",
     100            0 :         )));
     101           28 :     }
     102           28 : 
     103           28 :     fail::fail_point!("remote-storage-download-pre-rename", |_| {
     104            0 :         Err(DownloadError::Other(anyhow!(
     105            0 :             "remote-storage-download-pre-rename failpoint triggered"
     106            0 :         )))
     107           28 :     });
     108              : 
     109           28 :     fs::rename(&temp_file_path, &local_path)
     110           28 :         .await
     111           28 :         .with_context(|| format!("rename download layer file to {local_path}"))
     112           28 :         .map_err(DownloadError::Other)?;
     113              : 
     114              :     // We use fatal_err() below because the after the rename above,
     115              :     // the in-memory state of the filesystem already has the layer file in its final place,
     116              :     // and subsequent pageserver code could think it's durable while it really isn't.
     117           28 :     let work = {
     118           28 :         let ctx = ctx.detached_child(ctx.task_kind(), ctx.download_behavior());
     119           28 :         async move {
     120           28 :             let timeline_dir = VirtualFile::open(&timeline_path, &ctx)
     121           28 :                 .await
     122           28 :                 .fatal_err("VirtualFile::open for timeline dir fsync");
     123           28 :             timeline_dir
     124           28 :                 .sync_all()
     125           28 :                 .await
     126           28 :                 .fatal_err("VirtualFile::sync_all timeline dir");
     127           28 :         }
     128              :     };
     129           28 :     crate::virtual_file::io_engine::get()
     130           28 :         .spawn_blocking_and_block_on_if_std(work)
     131           28 :         .await;
     132              : 
     133           28 :     tracing::debug!("download complete: {local_path}");
     134              : 
     135           28 :     Ok(bytes_amount)
     136           28 : }
     137              : 
     138              : /// Download the object `src_path` in the remote `storage` to local path `dst_path`.
     139              : ///
     140              : /// If Ok() is returned, the download succeeded and the inode & data have been made durable.
     141              : /// (Note that the directory entry for the inode is not made durable.)
     142              : /// The file size in bytes is returned.
     143              : ///
     144              : /// If Err() is returned, there was some error. The file at `dst_path` has been unlinked.
     145              : /// The unlinking has _not_ been made durable.
     146           28 : async fn download_object(
     147           28 :     storage: &GenericRemoteStorage,
     148           28 :     src_path: &RemotePath,
     149           28 :     dst_path: &Utf8PathBuf,
     150           28 :     #[cfg_attr(target_os = "macos", allow(unused_variables))] gate: &utils::sync::gate::Gate,
     151           28 :     cancel: &CancellationToken,
     152           28 :     #[cfg_attr(target_os = "macos", allow(unused_variables))] ctx: &RequestContext,
     153           28 : ) -> Result<u64, DownloadError> {
     154           28 :     let res = match crate::virtual_file::io_engine::get() {
     155            0 :         crate::virtual_file::io_engine::IoEngine::NotSet => panic!("unset"),
     156              :         crate::virtual_file::io_engine::IoEngine::StdFs => {
     157           14 :             async {
     158           14 :                 let destination_file = tokio::fs::File::create(dst_path)
     159           14 :                     .await
     160           14 :                     .with_context(|| format!("create a destination file for layer '{dst_path}'"))
     161           14 :                     .map_err(DownloadError::Other)?;
     162              : 
     163           14 :                 let download = storage
     164           14 :                     .download(src_path, &DownloadOpts::default(), cancel)
     165           14 :                     .await?;
     166              : 
     167           14 :                 pausable_failpoint!("before-downloading-layer-stream-pausable");
     168              : 
     169           14 :                 let mut buf_writer =
     170           14 :                     tokio::io::BufWriter::with_capacity(super::BUFFER_SIZE, destination_file);
     171           14 : 
     172           14 :                 let mut reader = tokio_util::io::StreamReader::new(download.download_stream);
     173              : 
     174           14 :                 let bytes_amount = tokio::io::copy_buf(&mut reader, &mut buf_writer).await?;
     175           14 :                 buf_writer.flush().await?;
     176              : 
     177           14 :                 let mut destination_file = buf_writer.into_inner();
     178           14 : 
     179           14 :                 // Tokio doc here: https://docs.rs/tokio/1.17.0/tokio/fs/struct.File.html states that:
     180           14 :                 // A file will not be closed immediately when it goes out of scope if there are any IO operations
     181           14 :                 // that have not yet completed. To ensure that a file is closed immediately when it is dropped,
     182           14 :                 // you should call flush before dropping it.
     183           14 :                 //
     184           14 :                 // From the tokio code I see that it waits for pending operations to complete. There shouldt be any because
     185           14 :                 // we assume that `destination_file` file is fully written. I e there is no pending .write(...).await operations.
     186           14 :                 // But for additional safety lets check/wait for any pending operations.
     187           14 :                 destination_file
     188           14 :                     .flush()
     189           14 :                     .await
     190           14 :                     .maybe_fatal_err("download_object sync_all")
     191           14 :                     .with_context(|| format!("flush source file at {dst_path}"))
     192           14 :                     .map_err(DownloadError::Other)?;
     193              : 
     194              :                 // not using sync_data because it can lose file size update
     195           14 :                 destination_file
     196           14 :                     .sync_all()
     197           14 :                     .await
     198           14 :                     .maybe_fatal_err("download_object sync_all")
     199           14 :                     .with_context(|| format!("failed to fsync source file at {dst_path}"))
     200           14 :                     .map_err(DownloadError::Other)?;
     201              : 
     202           14 :                 Ok(bytes_amount)
     203           14 :             }
     204           14 :             .await
     205              :         }
     206              :         #[cfg(target_os = "linux")]
     207              :         crate::virtual_file::io_engine::IoEngine::TokioEpollUring => {
     208              :             use std::sync::Arc;
     209              : 
     210              :             use crate::virtual_file::{IoBufferMut, owned_buffers_io};
     211           14 :             async {
     212           14 :                 let destination_file = Arc::new(
     213           14 :                     VirtualFile::create(dst_path, ctx)
     214           14 :                         .await
     215           14 :                         .with_context(|| {
     216            0 :                             format!("create a destination file for layer '{dst_path}'")
     217           14 :                         })
     218           14 :                         .map_err(DownloadError::Other)?,
     219              :                 );
     220              : 
     221           14 :                 let mut download = storage
     222           14 :                     .download(src_path, &DownloadOpts::default(), cancel)
     223           14 :                     .await?;
     224              : 
     225           14 :                 pausable_failpoint!("before-downloading-layer-stream-pausable");
     226              : 
     227           14 :                 let mut buffered = owned_buffers_io::write::BufferedWriter::<IoBufferMut, _>::new(
     228           14 :                     destination_file,
     229           28 :                     || IoBufferMut::with_capacity(super::BUFFER_SIZE),
     230           14 :                     gate.enter().map_err(|_| DownloadError::Cancelled)?,
     231           14 :                     ctx,
     232           14 :                     info_span!(parent: None, "download_object_buffered_writer", %dst_path),
     233              :                 );
     234              : 
     235              :                 // TODO: use vectored write (writev) once supported by tokio-epoll-uring.
     236              :                 // There's chunks_vectored() on the stream.
     237           14 :                 let (bytes_amount, destination_file) = async {
     238           84 :                     while let Some(res) =
     239           98 :                         futures::StreamExt::next(&mut download.download_stream).await
     240              :                     {
     241           84 :                         let chunk = match res {
     242           84 :                             Ok(chunk) => chunk,
     243            0 :                             Err(e) => return Err(e),
     244              :                         };
     245           84 :                         buffered.write_buffered_borrowed(&chunk, ctx).await?;
     246              :                     }
     247           14 :                     let inner = buffered.flush_and_into_inner(ctx).await?;
     248           14 :                     Ok(inner)
     249           14 :                 }
     250           14 :                 .await?;
     251              : 
     252              :                 // not using sync_data because it can lose file size update
     253           14 :                 destination_file
     254           14 :                     .sync_all()
     255           14 :                     .await
     256           14 :                     .maybe_fatal_err("download_object sync_all")
     257           14 :                     .with_context(|| format!("failed to fsync source file at {dst_path}"))
     258           14 :                     .map_err(DownloadError::Other)?;
     259              : 
     260           14 :                 Ok(bytes_amount)
     261           14 :             }
     262           14 :             .await
     263              :         }
     264              :     };
     265              : 
     266              :     // in case the download failed, clean up
     267           28 :     match res {
     268           28 :         Ok(bytes_amount) => Ok(bytes_amount),
     269            0 :         Err(e) => {
     270            0 :             if let Err(e) = tokio::fs::remove_file(dst_path).await {
     271            0 :                 if e.kind() != std::io::ErrorKind::NotFound {
     272            0 :                     on_fatal_io_error(&e, &format!("Removing temporary file {dst_path}"));
     273            0 :                 }
     274            0 :             }
     275            0 :             Err(e)
     276              :         }
     277              :     }
     278           28 : }
     279              : 
     280              : const TEMP_DOWNLOAD_EXTENSION: &str = "temp_download";
     281              : 
     282            0 : pub(crate) fn is_temp_download_file(path: &Utf8Path) -> bool {
     283            0 :     let extension = path.extension();
     284            0 :     match extension {
     285            0 :         Some(TEMP_DOWNLOAD_EXTENSION) => true,
     286            0 :         Some(_) => false,
     287            0 :         None => false,
     288              :     }
     289            0 : }
     290              : 
     291          452 : async fn list_identifiers<T>(
     292          452 :     storage: &GenericRemoteStorage,
     293          452 :     prefix: RemotePath,
     294          452 :     cancel: CancellationToken,
     295          452 : ) -> anyhow::Result<(HashSet<T>, HashSet<String>)>
     296          452 : where
     297          452 :     T: FromStr + Eq + std::hash::Hash,
     298          452 : {
     299          452 :     let listing = download_retry_forever(
     300          452 :         || storage.list(Some(&prefix), ListingMode::WithDelimiter, None, &cancel),
     301          452 :         &format!("list identifiers in prefix {prefix}"),
     302          452 :         &cancel,
     303          452 :     )
     304          452 :     .await?;
     305              : 
     306          452 :     let mut parsed_ids = HashSet::new();
     307          452 :     let mut other_prefixes = HashSet::new();
     308              : 
     309          464 :     for id_remote_storage_key in listing.prefixes {
     310           12 :         let object_name = id_remote_storage_key.object_name().ok_or_else(|| {
     311            0 :             anyhow::anyhow!("failed to get object name for key {id_remote_storage_key}")
     312           12 :         })?;
     313              : 
     314           12 :         match object_name.parse::<T>() {
     315           12 :             Ok(t) => parsed_ids.insert(t),
     316            0 :             Err(_) => other_prefixes.insert(object_name.to_string()),
     317              :         };
     318              :     }
     319              : 
     320          452 :     for object in listing.keys {
     321            0 :         let object_name = object
     322            0 :             .key
     323            0 :             .object_name()
     324            0 :             .ok_or_else(|| anyhow::anyhow!("object name for key {}", object.key))?;
     325            0 :         other_prefixes.insert(object_name.to_string());
     326              :     }
     327              : 
     328          452 :     Ok((parsed_ids, other_prefixes))
     329          452 : }
     330              : 
     331              : /// List shards of given tenant in remote storage
     332            0 : pub(crate) async fn list_remote_tenant_shards(
     333            0 :     storage: &GenericRemoteStorage,
     334            0 :     tenant_id: TenantId,
     335            0 :     cancel: CancellationToken,
     336            0 : ) -> anyhow::Result<(HashSet<TenantShardId>, HashSet<String>)> {
     337            0 :     let remote_path = remote_tenant_path(&TenantShardId::unsharded(tenant_id));
     338            0 :     list_identifiers::<TenantShardId>(storage, remote_path, cancel).await
     339            0 : }
     340              : 
     341              : /// List timelines of given tenant shard in remote storage
     342          452 : pub async fn list_remote_timelines(
     343          452 :     storage: &GenericRemoteStorage,
     344          452 :     tenant_shard_id: TenantShardId,
     345          452 :     cancel: CancellationToken,
     346          452 : ) -> anyhow::Result<(HashSet<TimelineId>, HashSet<String>)> {
     347          452 :     fail::fail_point!("storage-sync-list-remote-timelines", |_| {
     348            0 :         anyhow::bail!("storage-sync-list-remote-timelines");
     349          452 :     });
     350              : 
     351          452 :     let remote_path = remote_timelines_path(&tenant_shard_id).add_trailing_slash();
     352          452 :     list_identifiers::<TimelineId>(storage, remote_path, cancel).await
     353          452 : }
     354              : 
     355         1424 : async fn do_download_remote_path_retry_forever(
     356         1424 :     storage: &GenericRemoteStorage,
     357         1424 :     remote_path: &RemotePath,
     358         1424 :     download_opts: DownloadOpts,
     359         1424 :     cancel: &CancellationToken,
     360         1424 : ) -> Result<(Vec<u8>, SystemTime), DownloadError> {
     361         1424 :     download_retry_forever(
     362         1424 :         || async {
     363         1424 :             let download = storage
     364         1424 :                 .download(remote_path, &download_opts, cancel)
     365         1424 :                 .await?;
     366              : 
     367           40 :             let mut bytes = Vec::new();
     368           40 : 
     369           40 :             let stream = download.download_stream;
     370           40 :             let mut stream = StreamReader::new(stream);
     371           40 : 
     372           40 :             tokio::io::copy_buf(&mut stream, &mut bytes).await?;
     373              : 
     374           40 :             Ok((bytes, download.last_modified))
     375         2848 :         },
     376         1424 :         &format!("download {remote_path:?}"),
     377         1424 :         cancel,
     378         1424 :     )
     379         1424 :     .await
     380         1424 : }
     381              : 
     382         1356 : async fn do_download_tenant_manifest(
     383         1356 :     storage: &GenericRemoteStorage,
     384         1356 :     tenant_shard_id: &TenantShardId,
     385         1356 :     _timeline_id: Option<&TimelineId>,
     386         1356 :     generation: Generation,
     387         1356 :     cancel: &CancellationToken,
     388         1356 : ) -> Result<(TenantManifest, Generation, SystemTime), DownloadError> {
     389         1356 :     let remote_path = remote_tenant_manifest_path(tenant_shard_id, generation);
     390         1356 : 
     391         1356 :     let download_opts = DownloadOpts {
     392         1356 :         kind: DownloadKind::Small,
     393         1356 :         ..Default::default()
     394         1356 :     };
     395              : 
     396            0 :     let (manifest_bytes, manifest_bytes_mtime) =
     397         1356 :         do_download_remote_path_retry_forever(storage, &remote_path, download_opts, cancel).await?;
     398              : 
     399            0 :     let tenant_manifest = TenantManifest::from_json_bytes(&manifest_bytes)
     400            0 :         .with_context(|| format!("deserialize tenant manifest file at {remote_path:?}"))
     401            0 :         .map_err(DownloadError::Other)?;
     402              : 
     403            0 :     Ok((tenant_manifest, generation, manifest_bytes_mtime))
     404         1356 : }
     405              : 
     406           68 : async fn do_download_index_part(
     407           68 :     storage: &GenericRemoteStorage,
     408           68 :     tenant_shard_id: &TenantShardId,
     409           68 :     timeline_id: Option<&TimelineId>,
     410           68 :     index_generation: Generation,
     411           68 :     cancel: &CancellationToken,
     412           68 : ) -> Result<(IndexPart, Generation, SystemTime), DownloadError> {
     413           68 :     let timeline_id =
     414           68 :         timeline_id.expect("A timeline ID is always provided when downloading an index");
     415           68 :     let remote_path = remote_index_path(tenant_shard_id, timeline_id, index_generation);
     416           68 : 
     417           68 :     let download_opts = DownloadOpts {
     418           68 :         kind: DownloadKind::Small,
     419           68 :         ..Default::default()
     420           68 :     };
     421              : 
     422           40 :     let (index_part_bytes, index_part_mtime) =
     423           68 :         do_download_remote_path_retry_forever(storage, &remote_path, download_opts, cancel).await?;
     424              : 
     425           40 :     let index_part: IndexPart = serde_json::from_slice(&index_part_bytes)
     426           40 :         .with_context(|| format!("deserialize index part file at {remote_path:?}"))
     427           40 :         .map_err(DownloadError::Other)?;
     428              : 
     429           40 :     Ok((index_part, index_generation, index_part_mtime))
     430           68 : }
     431              : 
     432              : /// Metadata objects are "generationed", meaning that they include a generation suffix.  This
     433              : /// function downloads the object with the highest generation <= `my_generation`.
     434              : ///
     435              : /// Data objects (layer files) also include a generation in their path, but there is no equivalent
     436              : /// search process, because their reference from an index includes the generation.
     437              : ///
     438              : /// An expensive object listing operation is only done if necessary: the typical fast path is to issue two
     439              : /// GET operations, one to our own generation (stale attachment case), and one to the immediately preceding
     440              : /// generation (normal case when migrating/restarting).  Only if both of these return 404 do we fall back
     441              : /// to listing objects.
     442              : ///
     443              : /// * `my_generation`: the value of `[crate::tenant::Tenant::generation]`
     444              : /// * `what`: for logging, what object are we downloading
     445              : /// * `prefix`: when listing objects, use this prefix (i.e. the part of the object path before the generation)
     446              : /// * `do_download`: a GET of the object in a particular generation, which should **retry indefinitely** unless
     447              : ///                  `cancel`` has fired.  This function does not do its own retries of GET operations, and relies
     448              : ///                  on the function passed in to do so.
     449              : /// * `parse_path`: parse a fully qualified remote storage path to get the generation of the object.
     450              : #[allow(clippy::too_many_arguments)]
     451              : #[tracing::instrument(skip_all, fields(generation=?my_generation))]
     452              : pub(crate) async fn download_generation_object<'a, T, DF, DFF, PF>(
     453              :     storage: &'a GenericRemoteStorage,
     454              :     tenant_shard_id: &'a TenantShardId,
     455              :     timeline_id: Option<&'a TimelineId>,
     456              :     my_generation: Generation,
     457              :     what: &str,
     458              :     prefix: RemotePath,
     459              :     do_download: DF,
     460              :     parse_path: PF,
     461              :     cancel: &'a CancellationToken,
     462              : ) -> Result<(T, Generation, SystemTime), DownloadError>
     463              : where
     464              :     DF: Fn(
     465              :         &'a GenericRemoteStorage,
     466              :         &'a TenantShardId,
     467              :         Option<&'a TimelineId>,
     468              :         Generation,
     469              :         &'a CancellationToken,
     470              :     ) -> DFF,
     471              :     DFF: Future<Output = Result<(T, Generation, SystemTime), DownloadError>>,
     472              :     PF: Fn(RemotePath) -> Option<Generation>,
     473              :     T: 'static,
     474              : {
     475              :     debug_assert_current_span_has_tenant_id();
     476              : 
     477              :     if my_generation.is_none() {
     478              :         // Operating without generations: just fetch the generation-less path
     479              :         return do_download(storage, tenant_shard_id, timeline_id, my_generation, cancel).await;
     480              :     }
     481              : 
     482              :     // Stale case: If we were intentionally attached in a stale generation, the remote object may already
     483              :     // exist in our generation.
     484              :     //
     485              :     // This is an optimization to avoid doing the listing for the general case below.
     486              :     let res = do_download(storage, tenant_shard_id, timeline_id, my_generation, cancel).await;
     487              :     match res {
     488              :         Ok(decoded) => {
     489              :             tracing::debug!("Found {what} from current generation (this is a stale attachment)");
     490              :             return Ok(decoded);
     491              :         }
     492              :         Err(DownloadError::NotFound) => {}
     493              :         Err(e) => return Err(e),
     494              :     };
     495              : 
     496              :     // Typical case: the previous generation of this tenant was running healthily, and had uploaded the object
     497              :     // we are seeking in that generation.  We may safely start from this index without doing a listing, because:
     498              :     //  - We checked for current generation case above
     499              :     //  - generations > my_generation are to be ignored
     500              :     //  - any other objects that exist would have an older generation than `previous_gen`, and
     501              :     //    we want to find the most recent object from a previous generation.
     502              :     //
     503              :     // This is an optimization to avoid doing the listing for the general case below.
     504              :     let res = do_download(
     505              :         storage,
     506              :         tenant_shard_id,
     507              :         timeline_id,
     508              :         my_generation.previous(),
     509              :         cancel,
     510              :     )
     511              :     .await;
     512              :     match res {
     513              :         Ok(decoded) => {
     514              :             tracing::debug!("Found {what} from previous generation");
     515              :             return Ok(decoded);
     516              :         }
     517              :         Err(DownloadError::NotFound) => {
     518              :             tracing::debug!("No {what} found from previous generation, falling back to listing");
     519              :         }
     520              :         Err(e) => {
     521              :             return Err(e);
     522              :         }
     523              :     }
     524              : 
     525              :     // General case/fallback: if there is no index at my_generation or prev_generation, then list all index_part.json
     526              :     // objects, and select the highest one with a generation <= my_generation.  Constructing the prefix is equivalent
     527              :     // to constructing a full index path with no generation, because the generation is a suffix.
     528              :     let paths = download_retry(
     529          464 :         || async {
     530          464 :             storage
     531          464 :                 .list(Some(&prefix), ListingMode::NoDelimiter, None, cancel)
     532          464 :                 .await
     533          928 :         },
     534              :         "list index_part files",
     535              :         cancel,
     536              :     )
     537              :     .await?
     538              :     .keys;
     539              : 
     540              :     // General case logic for which index to use: the latest index whose generation
     541              :     // is <= our own.  See "Finding the remote indices for timelines" in docs/rfcs/025-generation-numbers.md
     542              :     let max_previous_generation = paths
     543              :         .into_iter()
     544           36 :         .filter_map(|o| parse_path(o.key))
     545           24 :         .filter(|g| g <= &my_generation)
     546              :         .max();
     547              : 
     548              :     match max_previous_generation {
     549              :         Some(g) => {
     550              :             tracing::debug!("Found {what} in generation {g:?}");
     551              :             do_download(storage, tenant_shard_id, timeline_id, g, cancel).await
     552              :         }
     553              :         None => {
     554              :             // Migration from legacy pre-generation state: we have a generation but no prior
     555              :             // attached pageservers did.  Try to load from a no-generation path.
     556              :             tracing::debug!("No {what}* found");
     557              :             do_download(
     558              :                 storage,
     559              :                 tenant_shard_id,
     560              :                 timeline_id,
     561              :                 Generation::none(),
     562              :                 cancel,
     563              :             )
     564              :             .await
     565              :         }
     566              :     }
     567              : }
     568              : 
     569              : /// index_part.json objects are suffixed with a generation number, so we cannot
     570              : /// directly GET the latest index part without doing some probing.
     571              : ///
     572              : /// In this function we probe for the most recent index in a generation <= our current generation.
     573              : /// See "Finding the remote indices for timelines" in docs/rfcs/025-generation-numbers.md
     574           40 : pub(crate) async fn download_index_part(
     575           40 :     storage: &GenericRemoteStorage,
     576           40 :     tenant_shard_id: &TenantShardId,
     577           40 :     timeline_id: &TimelineId,
     578           40 :     my_generation: Generation,
     579           40 :     cancel: &CancellationToken,
     580           40 : ) -> Result<(IndexPart, Generation, SystemTime), DownloadError> {
     581           40 :     debug_assert_current_span_has_tenant_and_timeline_id();
     582           40 : 
     583           40 :     let index_prefix = remote_index_path(tenant_shard_id, timeline_id, Generation::none());
     584           40 :     download_generation_object(
     585           40 :         storage,
     586           40 :         tenant_shard_id,
     587           40 :         Some(timeline_id),
     588           40 :         my_generation,
     589           40 :         "index_part",
     590           40 :         index_prefix,
     591           40 :         do_download_index_part,
     592           40 :         parse_remote_index_path,
     593           40 :         cancel,
     594           40 :     )
     595           40 :     .await
     596           40 : }
     597              : 
     598          452 : pub(crate) async fn download_tenant_manifest(
     599          452 :     storage: &GenericRemoteStorage,
     600          452 :     tenant_shard_id: &TenantShardId,
     601          452 :     my_generation: Generation,
     602          452 :     cancel: &CancellationToken,
     603          452 : ) -> Result<(TenantManifest, Generation, SystemTime), DownloadError> {
     604          452 :     let manifest_prefix = remote_tenant_manifest_prefix(tenant_shard_id);
     605          452 : 
     606          452 :     download_generation_object(
     607          452 :         storage,
     608          452 :         tenant_shard_id,
     609          452 :         None,
     610          452 :         my_generation,
     611          452 :         "tenant-manifest",
     612          452 :         manifest_prefix,
     613          452 :         do_download_tenant_manifest,
     614          452 :         parse_remote_tenant_manifest_path,
     615          452 :         cancel,
     616          452 :     )
     617          452 :     .await
     618          452 : }
     619              : 
     620            4 : pub(crate) async fn download_initdb_tar_zst(
     621            4 :     conf: &'static PageServerConf,
     622            4 :     storage: &GenericRemoteStorage,
     623            4 :     tenant_shard_id: &TenantShardId,
     624            4 :     timeline_id: &TimelineId,
     625            4 :     cancel: &CancellationToken,
     626            4 : ) -> Result<(Utf8PathBuf, File), DownloadError> {
     627            4 :     debug_assert_current_span_has_tenant_and_timeline_id();
     628            4 : 
     629            4 :     let remote_path = remote_initdb_archive_path(&tenant_shard_id.tenant_id, timeline_id);
     630            4 : 
     631            4 :     let remote_preserved_path =
     632            4 :         remote_initdb_preserved_archive_path(&tenant_shard_id.tenant_id, timeline_id);
     633            4 : 
     634            4 :     let timeline_path = conf.timelines_path(tenant_shard_id);
     635            4 : 
     636            4 :     if !timeline_path.exists() {
     637            0 :         tokio::fs::create_dir_all(&timeline_path)
     638            0 :             .await
     639            0 :             .with_context(|| format!("timeline dir creation {timeline_path}"))
     640            0 :             .map_err(DownloadError::Other)?;
     641            4 :     }
     642            4 :     let temp_path = timeline_path.join(format!(
     643            4 :         "{INITDB_PATH}.download-{timeline_id}.{TEMP_FILE_SUFFIX}"
     644            4 :     ));
     645              : 
     646            4 :     let file = download_retry(
     647            4 :         || async {
     648            4 :             let file = OpenOptions::new()
     649            4 :                 .create(true)
     650            4 :                 .truncate(true)
     651            4 :                 .read(true)
     652            4 :                 .write(true)
     653            4 :                 .open(&temp_path)
     654            4 :                 .await
     655            4 :                 .with_context(|| format!("tempfile creation {temp_path}"))
     656            4 :                 .map_err(DownloadError::Other)?;
     657              : 
     658            4 :             let download = match storage
     659            4 :                 .download(&remote_path, &DownloadOpts::default(), cancel)
     660            4 :                 .await
     661              :             {
     662            4 :                 Ok(dl) => dl,
     663              :                 Err(DownloadError::NotFound) => {
     664            0 :                     storage
     665            0 :                         .download(&remote_preserved_path, &DownloadOpts::default(), cancel)
     666            0 :                         .await?
     667              :                 }
     668            0 :                 Err(other) => Err(other)?,
     669              :             };
     670            4 :             let mut download = tokio_util::io::StreamReader::new(download.download_stream);
     671            4 :             let mut writer = tokio::io::BufWriter::with_capacity(super::BUFFER_SIZE, file);
     672            4 : 
     673            4 :             tokio::io::copy_buf(&mut download, &mut writer).await?;
     674              : 
     675            4 :             let mut file = writer.into_inner();
     676            4 : 
     677            4 :             file.seek(std::io::SeekFrom::Start(0))
     678            4 :                 .await
     679            4 :                 .with_context(|| format!("rewinding initdb.tar.zst at: {remote_path:?}"))
     680            4 :                 .map_err(DownloadError::Other)?;
     681              : 
     682            4 :             Ok(file)
     683            8 :         },
     684            4 :         &format!("download {remote_path}"),
     685            4 :         cancel,
     686            4 :     )
     687            4 :     .await
     688            4 :     .inspect_err(|_e| {
     689              :         // Do a best-effort attempt at deleting the temporary file upon encountering an error.
     690              :         // We don't have async here nor do we want to pile on any extra errors.
     691            0 :         if let Err(e) = std::fs::remove_file(&temp_path) {
     692            0 :             if e.kind() != std::io::ErrorKind::NotFound {
     693            0 :                 warn!("error deleting temporary file {temp_path}: {e}");
     694            0 :             }
     695            0 :         }
     696            4 :     })?;
     697              : 
     698            4 :     Ok((temp_path, file))
     699            4 : }
     700              : 
     701              : /// Helper function to handle retries for a download operation.
     702              : ///
     703              : /// Remote operations can fail due to rate limits (S3), spurious network
     704              : /// problems, or other external reasons. Retry FAILED_DOWNLOAD_RETRIES times,
     705              : /// with backoff.
     706              : ///
     707              : /// (See similar logic for uploads in `perform_upload_task`)
     708          496 : pub(super) async fn download_retry<T, O, F>(
     709          496 :     op: O,
     710          496 :     description: &str,
     711          496 :     cancel: &CancellationToken,
     712          496 : ) -> Result<T, DownloadError>
     713          496 : where
     714          496 :     O: FnMut() -> F,
     715          496 :     F: Future<Output = Result<T, DownloadError>>,
     716          496 : {
     717          496 :     backoff::retry(
     718          496 :         op,
     719          496 :         DownloadError::is_permanent,
     720          496 :         FAILED_DOWNLOAD_WARN_THRESHOLD,
     721          496 :         FAILED_REMOTE_OP_RETRIES,
     722          496 :         description,
     723          496 :         cancel,
     724          496 :     )
     725          496 :     .await
     726          496 :     .ok_or_else(|| DownloadError::Cancelled)
     727          496 :     .and_then(|x| x)
     728          496 : }
     729              : 
     730         1876 : pub(crate) async fn download_retry_forever<T, O, F>(
     731         1876 :     op: O,
     732         1876 :     description: &str,
     733         1876 :     cancel: &CancellationToken,
     734         1876 : ) -> Result<T, DownloadError>
     735         1876 : where
     736         1876 :     O: FnMut() -> F,
     737         1876 :     F: Future<Output = Result<T, DownloadError>>,
     738         1876 : {
     739         1876 :     backoff::retry(
     740         1876 :         op,
     741         1876 :         DownloadError::is_permanent,
     742         1876 :         FAILED_DOWNLOAD_WARN_THRESHOLD,
     743         1876 :         u32::MAX,
     744         1876 :         description,
     745         1876 :         cancel,
     746         1876 :     )
     747         1876 :     .await
     748         1876 :     .ok_or_else(|| DownloadError::Cancelled)
     749         1876 :     .and_then(|x| x)
     750         1876 : }
        

Generated by: LCOV version 2.1-beta