LCOV - code coverage report
Current view: top level - pageserver/src/tenant/secondary - downloader.rs (source / functions) Coverage Total Hit
Test: 5fe7fa8d483b39476409aee736d6d5e32728bfac.info Lines: 0.0 % 951 0
Test Date: 2025-03-12 16:10:49 Functions: 0.0 % 76 0

            Line data    Source code
       1              : use std::collections::{HashMap, HashSet};
       2              : use std::pin::Pin;
       3              : use std::str::FromStr;
       4              : use std::sync::Arc;
       5              : use std::time::{Duration, Instant, SystemTime};
       6              : 
       7              : use camino::Utf8PathBuf;
       8              : use chrono::format::{DelayedFormat, StrftimeItems};
       9              : use futures::Future;
      10              : use metrics::UIntGauge;
      11              : use pageserver_api::models::SecondaryProgress;
      12              : use pageserver_api::shard::TenantShardId;
      13              : use remote_storage::{DownloadError, DownloadKind, DownloadOpts, Etag, GenericRemoteStorage};
      14              : use tokio_util::sync::CancellationToken;
      15              : use tracing::{Instrument, info_span, instrument, warn};
      16              : use utils::completion::Barrier;
      17              : use utils::crashsafe::path_with_suffix_extension;
      18              : use utils::id::TimelineId;
      19              : use utils::{backoff, failpoint_support, fs_ext, pausable_failpoint, serde_system_time};
      20              : 
      21              : use super::heatmap::{HeatMapLayer, HeatMapTenant, HeatMapTimeline};
      22              : use super::scheduler::{
      23              :     self, Completion, JobGenerator, SchedulingResult, TenantBackgroundJobs, period_jitter,
      24              :     period_warmup,
      25              : };
      26              : use super::{
      27              :     CommandRequest, DownloadCommand, GetTenantError, SecondaryTenant, SecondaryTenantError,
      28              : };
      29              : use crate::TEMP_FILE_SUFFIX;
      30              : use crate::config::PageServerConf;
      31              : use crate::context::RequestContext;
      32              : use crate::disk_usage_eviction_task::{
      33              :     DiskUsageEvictionInfo, EvictionCandidate, EvictionLayer, EvictionSecondaryLayer, finite_f32,
      34              : };
      35              : use crate::metrics::SECONDARY_MODE;
      36              : use crate::tenant::config::SecondaryLocationConfig;
      37              : use crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id;
      38              : use crate::tenant::ephemeral_file::is_ephemeral_file;
      39              : use crate::tenant::mgr::TenantManager;
      40              : use crate::tenant::remote_timeline_client::download::download_layer_file;
      41              : use crate::tenant::remote_timeline_client::index::LayerFileMetadata;
      42              : use crate::tenant::remote_timeline_client::{
      43              :     FAILED_DOWNLOAD_WARN_THRESHOLD, FAILED_REMOTE_OP_RETRIES, is_temp_download_file,
      44              :     remote_heatmap_path,
      45              : };
      46              : use crate::tenant::span::debug_assert_current_span_has_tenant_id;
      47              : use crate::tenant::storage_layer::layer::local_layer_path;
      48              : use crate::tenant::storage_layer::{LayerName, LayerVisibilityHint};
      49              : use crate::tenant::tasks::{BackgroundLoopKind, warn_when_period_overrun};
      50              : use crate::virtual_file::{MaybeFatalIo, VirtualFile, on_fatal_io_error};
      51              : 
      52              : /// For each tenant, default period for how long must have passed since the last download_tenant call before
      53              : /// calling it again.  This default is replaced with the value of [`HeatMapTenant::upload_period_ms`] after first
      54              : /// download, if the uploader populated it.
      55              : const DEFAULT_DOWNLOAD_INTERVAL: Duration = Duration::from_millis(60000);
      56              : 
      57            0 : pub(super) async fn downloader_task(
      58            0 :     tenant_manager: Arc<TenantManager>,
      59            0 :     remote_storage: GenericRemoteStorage,
      60            0 :     command_queue: tokio::sync::mpsc::Receiver<CommandRequest<DownloadCommand>>,
      61            0 :     background_jobs_can_start: Barrier,
      62            0 :     cancel: CancellationToken,
      63            0 :     root_ctx: RequestContext,
      64            0 : ) {
      65            0 :     let concurrency = tenant_manager.get_conf().secondary_download_concurrency;
      66            0 : 
      67            0 :     let generator = SecondaryDownloader {
      68            0 :         tenant_manager,
      69            0 :         remote_storage,
      70            0 :         root_ctx,
      71            0 :     };
      72            0 :     let mut scheduler = Scheduler::new(generator, concurrency);
      73            0 : 
      74            0 :     scheduler
      75            0 :         .run(command_queue, background_jobs_can_start, cancel)
      76            0 :         .instrument(info_span!("secondary_download_scheduler"))
      77            0 :         .await
      78            0 : }
      79              : 
      80              : struct SecondaryDownloader {
      81              :     tenant_manager: Arc<TenantManager>,
      82              :     remote_storage: GenericRemoteStorage,
      83              :     root_ctx: RequestContext,
      84              : }
      85              : 
      86              : #[derive(Debug, Clone)]
      87              : pub(super) struct OnDiskState {
      88              :     metadata: LayerFileMetadata,
      89              :     access_time: SystemTime,
      90              :     local_path: Utf8PathBuf,
      91              : }
      92              : 
      93              : impl OnDiskState {
      94            0 :     fn new(
      95            0 :         _conf: &'static PageServerConf,
      96            0 :         _tenant_shard_id: &TenantShardId,
      97            0 :         _imeline_id: &TimelineId,
      98            0 :         _ame: LayerName,
      99            0 :         metadata: LayerFileMetadata,
     100            0 :         access_time: SystemTime,
     101            0 :         local_path: Utf8PathBuf,
     102            0 :     ) -> Self {
     103            0 :         Self {
     104            0 :             metadata,
     105            0 :             access_time,
     106            0 :             local_path,
     107            0 :         }
     108            0 :     }
     109              : 
     110              :     // This is infallible, because all errors are either acceptable (ENOENT), or totally
     111              :     // unexpected (fatal).
     112            0 :     pub(super) fn remove_blocking(&self) {
     113            0 :         // We tolerate ENOENT, because between planning eviction and executing
     114            0 :         // it, the secondary downloader could have seen an updated heatmap that
     115            0 :         // resulted in a layer being deleted.
     116            0 :         // Other local I/O errors are process-fatal: these should never happen.
     117            0 :         std::fs::remove_file(&self.local_path)
     118            0 :             .or_else(fs_ext::ignore_not_found)
     119            0 :             .fatal_err("Deleting secondary layer")
     120            0 :     }
     121              : 
     122            0 :     pub(crate) fn file_size(&self) -> u64 {
     123            0 :         self.metadata.file_size
     124            0 :     }
     125              : }
     126              : 
     127              : #[derive(Debug, Clone, Default)]
     128              : pub(super) struct SecondaryDetailTimeline {
     129              :     on_disk_layers: HashMap<LayerName, OnDiskState>,
     130              : 
     131              :     /// We remember when layers were evicted, to prevent re-downloading them.
     132              :     pub(super) evicted_at: HashMap<LayerName, SystemTime>,
     133              : }
     134              : 
     135              : impl SecondaryDetailTimeline {
     136            0 :     pub(super) fn remove_layer(
     137            0 :         &mut self,
     138            0 :         name: &LayerName,
     139            0 :         resident_metric: &UIntGauge,
     140            0 :     ) -> Option<OnDiskState> {
     141            0 :         let removed = self.on_disk_layers.remove(name);
     142            0 :         if let Some(removed) = &removed {
     143            0 :             resident_metric.sub(removed.file_size());
     144            0 :         }
     145            0 :         removed
     146            0 :     }
     147              : 
     148              :     /// `local_path`
     149            0 :     fn touch_layer<F>(
     150            0 :         &mut self,
     151            0 :         conf: &'static PageServerConf,
     152            0 :         tenant_shard_id: &TenantShardId,
     153            0 :         timeline_id: &TimelineId,
     154            0 :         touched: &HeatMapLayer,
     155            0 :         resident_metric: &UIntGauge,
     156            0 :         local_path: F,
     157            0 :     ) where
     158            0 :         F: FnOnce() -> Utf8PathBuf,
     159            0 :     {
     160              :         use std::collections::hash_map::Entry;
     161            0 :         match self.on_disk_layers.entry(touched.name.clone()) {
     162            0 :             Entry::Occupied(mut v) => {
     163            0 :                 v.get_mut().access_time = touched.access_time;
     164            0 :             }
     165            0 :             Entry::Vacant(e) => {
     166            0 :                 e.insert(OnDiskState::new(
     167            0 :                     conf,
     168            0 :                     tenant_shard_id,
     169            0 :                     timeline_id,
     170            0 :                     touched.name.clone(),
     171            0 :                     touched.metadata.clone(),
     172            0 :                     touched.access_time,
     173            0 :                     local_path(),
     174            0 :                 ));
     175            0 :                 resident_metric.add(touched.metadata.file_size);
     176            0 :             }
     177              :         }
     178            0 :     }
     179              : }
     180              : 
     181              : // Aspects of a heatmap that we remember after downloading it
     182              : #[derive(Clone, Debug)]
     183              : struct DownloadSummary {
     184              :     etag: Etag,
     185              :     #[allow(unused)]
     186              :     mtime: SystemTime,
     187              :     upload_period: Duration,
     188              : }
     189              : 
     190              : /// This state is written by the secondary downloader, it is opaque
     191              : /// to TenantManager
     192              : #[derive(Debug)]
     193              : pub(super) struct SecondaryDetail {
     194              :     pub(super) config: SecondaryLocationConfig,
     195              : 
     196              :     last_download: Option<DownloadSummary>,
     197              :     next_download: Option<Instant>,
     198              :     timelines: HashMap<TimelineId, SecondaryDetailTimeline>,
     199              : }
     200              : 
     201              : /// Helper for logging SystemTime
     202            0 : fn strftime(t: &'_ SystemTime) -> DelayedFormat<StrftimeItems<'_>> {
     203            0 :     let datetime: chrono::DateTime<chrono::Utc> = (*t).into();
     204            0 :     datetime.format("%d/%m/%Y %T")
     205            0 : }
     206              : 
     207              : /// Information returned from download function when it detects the heatmap has changed
     208              : struct HeatMapModified {
     209              :     etag: Etag,
     210              :     last_modified: SystemTime,
     211              :     bytes: Vec<u8>,
     212              : }
     213              : 
     214              : enum HeatMapDownload {
     215              :     // The heatmap's etag has changed: return the new etag, mtime and the body bytes
     216              :     Modified(HeatMapModified),
     217              :     // The heatmap's etag is unchanged
     218              :     Unmodified,
     219              : }
     220              : 
     221              : impl SecondaryDetail {
     222            0 :     pub(super) fn new(config: SecondaryLocationConfig) -> Self {
     223            0 :         Self {
     224            0 :             config,
     225            0 :             last_download: None,
     226            0 :             next_download: None,
     227            0 :             timelines: HashMap::new(),
     228            0 :         }
     229            0 :     }
     230              : 
     231              :     #[cfg(feature = "testing")]
     232            0 :     pub(crate) fn total_resident_size(&self) -> u64 {
     233            0 :         self.timelines
     234            0 :             .values()
     235            0 :             .map(|tl| {
     236            0 :                 tl.on_disk_layers
     237            0 :                     .values()
     238            0 :                     .map(|v| v.metadata.file_size)
     239            0 :                     .sum::<u64>()
     240            0 :             })
     241            0 :             .sum::<u64>()
     242            0 :     }
     243              : 
     244            0 :     pub(super) fn evict_layer(
     245            0 :         &mut self,
     246            0 :         name: LayerName,
     247            0 :         timeline_id: &TimelineId,
     248            0 :         now: SystemTime,
     249            0 :         resident_metric: &UIntGauge,
     250            0 :     ) -> Option<OnDiskState> {
     251            0 :         let timeline = self.timelines.get_mut(timeline_id)?;
     252            0 :         let removed = timeline.remove_layer(&name, resident_metric);
     253            0 :         if removed.is_some() {
     254            0 :             timeline.evicted_at.insert(name, now);
     255            0 :         }
     256            0 :         removed
     257            0 :     }
     258              : 
     259            0 :     pub(super) fn remove_timeline(
     260            0 :         &mut self,
     261            0 :         timeline_id: &TimelineId,
     262            0 :         resident_metric: &UIntGauge,
     263            0 :     ) {
     264            0 :         let removed = self.timelines.remove(timeline_id);
     265            0 :         if let Some(removed) = removed {
     266            0 :             resident_metric.sub(
     267            0 :                 removed
     268            0 :                     .on_disk_layers
     269            0 :                     .values()
     270            0 :                     .map(|l| l.metadata.file_size)
     271            0 :                     .sum(),
     272            0 :             );
     273            0 :         }
     274            0 :     }
     275              : 
     276              :     /// Additionally returns the total number of layers, used for more stable relative access time
     277              :     /// based eviction.
     278            0 :     pub(super) fn get_layers_for_eviction(
     279            0 :         &self,
     280            0 :         parent: &Arc<SecondaryTenant>,
     281            0 :     ) -> (DiskUsageEvictionInfo, usize) {
     282            0 :         let mut result = DiskUsageEvictionInfo::default();
     283            0 :         let mut total_layers = 0;
     284              : 
     285            0 :         for (timeline_id, timeline_detail) in &self.timelines {
     286            0 :             result
     287            0 :                 .resident_layers
     288            0 :                 .extend(timeline_detail.on_disk_layers.iter().map(|(name, ods)| {
     289            0 :                     EvictionCandidate {
     290            0 :                         layer: EvictionLayer::Secondary(EvictionSecondaryLayer {
     291            0 :                             secondary_tenant: parent.clone(),
     292            0 :                             timeline_id: *timeline_id,
     293            0 :                             name: name.clone(),
     294            0 :                             metadata: ods.metadata.clone(),
     295            0 :                         }),
     296            0 :                         last_activity_ts: ods.access_time,
     297            0 :                         relative_last_activity: finite_f32::FiniteF32::ZERO,
     298            0 :                         // Secondary location layers are presumed visible, because Covered layers
     299            0 :                         // are excluded from the heatmap
     300            0 :                         visibility: LayerVisibilityHint::Visible,
     301            0 :                     }
     302            0 :                 }));
     303            0 : 
     304            0 :             // total might be missing currently downloading layers, but as a lower than actual
     305            0 :             // value it is good enough approximation.
     306            0 :             total_layers += timeline_detail.on_disk_layers.len() + timeline_detail.evicted_at.len();
     307            0 :         }
     308            0 :         result.max_layer_size = result
     309            0 :             .resident_layers
     310            0 :             .iter()
     311            0 :             .map(|l| l.layer.get_file_size())
     312            0 :             .max();
     313            0 : 
     314            0 :         tracing::debug!(
     315            0 :             "eviction: secondary tenant {} found {} timelines, {} layers",
     316            0 :             parent.get_tenant_shard_id(),
     317            0 :             self.timelines.len(),
     318            0 :             result.resident_layers.len()
     319              :         );
     320              : 
     321            0 :         (result, total_layers)
     322            0 :     }
     323              : }
     324              : 
     325              : struct PendingDownload {
     326              :     secondary_state: Arc<SecondaryTenant>,
     327              :     last_download: Option<DownloadSummary>,
     328              :     target_time: Option<Instant>,
     329              : }
     330              : 
     331              : impl scheduler::PendingJob for PendingDownload {
     332            0 :     fn get_tenant_shard_id(&self) -> &TenantShardId {
     333            0 :         self.secondary_state.get_tenant_shard_id()
     334            0 :     }
     335              : }
     336              : 
     337              : struct RunningDownload {
     338              :     barrier: Barrier,
     339              : }
     340              : 
     341              : impl scheduler::RunningJob for RunningDownload {
     342            0 :     fn get_barrier(&self) -> Barrier {
     343            0 :         self.barrier.clone()
     344            0 :     }
     345              : }
     346              : 
     347              : struct CompleteDownload {
     348              :     secondary_state: Arc<SecondaryTenant>,
     349              :     completed_at: Instant,
     350              :     result: Result<(), UpdateError>,
     351              : }
     352              : 
     353              : impl scheduler::Completion for CompleteDownload {
     354            0 :     fn get_tenant_shard_id(&self) -> &TenantShardId {
     355            0 :         self.secondary_state.get_tenant_shard_id()
     356            0 :     }
     357              : }
     358              : 
     359              : type Scheduler = TenantBackgroundJobs<
     360              :     SecondaryDownloader,
     361              :     PendingDownload,
     362              :     RunningDownload,
     363              :     CompleteDownload,
     364              :     DownloadCommand,
     365              : >;
     366              : 
     367              : impl JobGenerator<PendingDownload, RunningDownload, CompleteDownload, DownloadCommand>
     368              :     for SecondaryDownloader
     369              : {
     370              :     #[instrument(skip_all, fields(tenant_id=%completion.get_tenant_shard_id().tenant_id, shard_id=%completion.get_tenant_shard_id().shard_slug()))]
     371              :     fn on_completion(&mut self, completion: CompleteDownload) {
     372              :         let CompleteDownload {
     373              :             secondary_state,
     374              :             completed_at: _completed_at,
     375              :             result,
     376              :         } = completion;
     377              : 
     378              :         tracing::debug!("Secondary tenant download completed");
     379              : 
     380              :         let mut detail = secondary_state.detail.lock().unwrap();
     381              : 
     382              :         match result {
     383              :             Err(UpdateError::Restart) => {
     384              :                 // Start downloading again as soon as we can.  This will involve waiting for the scheduler's
     385              :                 // scheduling interval.  This slightly reduces the peak download speed of tenants that hit their
     386              :                 // deadline and keep restarting, but that also helps give other tenants a chance to execute rather
     387              :                 // that letting one big tenant dominate for a long time.
     388              :                 detail.next_download = Some(Instant::now());
     389              :             }
     390              :             _ => {
     391              :                 let period = detail
     392              :                     .last_download
     393              :                     .as_ref()
     394            0 :                     .map(|d| d.upload_period)
     395              :                     .unwrap_or(DEFAULT_DOWNLOAD_INTERVAL);
     396              : 
     397              :                 // We advance next_download irrespective of errors: we don't want error cases to result in
     398              :                 // expensive busy-polling.
     399              :                 detail.next_download = Some(Instant::now() + period_jitter(period, 5));
     400              :             }
     401              :         }
     402              :     }
     403              : 
     404            0 :     async fn schedule(&mut self) -> SchedulingResult<PendingDownload> {
     405            0 :         let mut result = SchedulingResult {
     406            0 :             jobs: Vec::new(),
     407            0 :             want_interval: None,
     408            0 :         };
     409            0 : 
     410            0 :         // Step 1: identify some tenants that we may work on
     411            0 :         let mut tenants: Vec<Arc<SecondaryTenant>> = Vec::new();
     412            0 :         self.tenant_manager
     413            0 :             .foreach_secondary_tenants(|_id, secondary_state| {
     414            0 :                 tenants.push(secondary_state.clone());
     415            0 :             });
     416            0 : 
     417            0 :         // Step 2: filter out tenants which are not yet elegible to run
     418            0 :         let now = Instant::now();
     419            0 :         result.jobs = tenants
     420            0 :             .into_iter()
     421            0 :             .filter_map(|secondary_tenant| {
     422            0 :                 let (last_download, next_download) = {
     423            0 :                     let mut detail = secondary_tenant.detail.lock().unwrap();
     424            0 : 
     425            0 :                     if !detail.config.warm {
     426              :                         // Downloads are disabled for this tenant
     427            0 :                         detail.next_download = None;
     428            0 :                         return None;
     429            0 :                     }
     430            0 : 
     431            0 :                     if detail.next_download.is_none() {
     432            0 :                         // Initialize randomly in the range from 0 to our interval: this uniformly spreads the start times.  Subsequent
     433            0 :                         // rounds will use a smaller jitter to avoid accidentally synchronizing later.
     434            0 :                         detail.next_download = Some(now.checked_add(period_warmup(DEFAULT_DOWNLOAD_INTERVAL)).expect(
     435            0 :                         "Using our constant, which is known to be small compared with clock range",
     436            0 :                     ));
     437            0 :                     }
     438            0 :                     (detail.last_download.clone(), detail.next_download.unwrap())
     439            0 :                 };
     440            0 : 
     441            0 :                 if now > next_download {
     442            0 :                     Some(PendingDownload {
     443            0 :                         secondary_state: secondary_tenant,
     444            0 :                         last_download,
     445            0 :                         target_time: Some(next_download),
     446            0 :                     })
     447              :                 } else {
     448            0 :                     None
     449              :                 }
     450            0 :             })
     451            0 :             .collect();
     452            0 : 
     453            0 :         // Step 3: sort by target execution time to run most urgent first.
     454            0 :         result.jobs.sort_by_key(|j| j.target_time);
     455            0 : 
     456            0 :         result
     457            0 :     }
     458              : 
     459            0 :     fn on_command(
     460            0 :         &mut self,
     461            0 :         command: DownloadCommand,
     462            0 :     ) -> Result<PendingDownload, SecondaryTenantError> {
     463            0 :         let tenant_shard_id = command.get_tenant_shard_id();
     464              : 
     465            0 :         let tenant = self
     466            0 :             .tenant_manager
     467            0 :             .get_secondary_tenant_shard(*tenant_shard_id)
     468            0 :             .ok_or(GetTenantError::ShardNotFound(*tenant_shard_id))?;
     469              : 
     470            0 :         Ok(PendingDownload {
     471            0 :             target_time: None,
     472            0 :             last_download: None,
     473            0 :             secondary_state: tenant,
     474            0 :         })
     475            0 :     }
     476              : 
     477            0 :     fn spawn(
     478            0 :         &mut self,
     479            0 :         job: PendingDownload,
     480            0 :     ) -> (
     481            0 :         RunningDownload,
     482            0 :         Pin<Box<dyn Future<Output = CompleteDownload> + Send>>,
     483            0 :     ) {
     484            0 :         let PendingDownload {
     485            0 :             secondary_state,
     486            0 :             last_download,
     487            0 :             target_time,
     488            0 :         } = job;
     489            0 : 
     490            0 :         let (completion, barrier) = utils::completion::channel();
     491            0 :         let remote_storage = self.remote_storage.clone();
     492            0 :         let conf = self.tenant_manager.get_conf();
     493            0 :         let tenant_shard_id = *secondary_state.get_tenant_shard_id();
     494            0 :         let download_ctx = self
     495            0 :             .root_ctx
     496            0 :             .attached_child()
     497            0 :             .with_scope_secondary_tenant(&tenant_shard_id);
     498            0 :         (RunningDownload { barrier }, Box::pin(async move {
     499            0 :             let _completion = completion;
     500              : 
     501            0 :             let result = TenantDownloader::new(conf, &remote_storage, &secondary_state)
     502            0 :                 .download(&download_ctx)
     503            0 :                 .await;
     504            0 :             match &result
     505              :             {
     506              :                 Err(UpdateError::NoData) => {
     507            0 :                     tracing::info!("No heatmap found for tenant.  This is fine if it is new.");
     508              :                 },
     509              :                 Err(UpdateError::NoSpace) => {
     510            0 :                     tracing::warn!("Insufficient space while downloading.  Will retry later.");
     511              :                 }
     512              :                 Err(UpdateError::Cancelled) => {
     513            0 :                     tracing::info!("Shut down while downloading");
     514              :                 },
     515            0 :                 Err(UpdateError::Deserialize(e)) => {
     516            0 :                     tracing::error!("Corrupt content while downloading tenant: {e}");
     517              :                 },
     518            0 :                 Err(e @ (UpdateError::DownloadError(_) | UpdateError::Other(_))) => {
     519            0 :                     tracing::error!("Error while downloading tenant: {e}");
     520              :                 },
     521              :                 Err(UpdateError::Restart) => {
     522            0 :                     tracing::info!("Download reached deadline & will restart to update heatmap")
     523              :                 }
     524            0 :                 Ok(()) => {}
     525              :             };
     526              : 
     527              :             // Irrespective of the result, we will reschedule ourselves to run after our usual period.
     528              : 
     529              :             // If the job had a target execution time, we may check our final execution
     530              :             // time against that for observability purposes.
     531            0 :             if let (Some(target_time), Some(last_download)) = (target_time, last_download) {
     532            0 :                 // Elapsed time includes any scheduling lag as well as the execution of the job
     533            0 :                 let elapsed = Instant::now().duration_since(target_time);
     534            0 : 
     535            0 :                 warn_when_period_overrun(
     536            0 :                     elapsed,
     537            0 :                     last_download.upload_period,
     538            0 :                     BackgroundLoopKind::SecondaryDownload,
     539            0 :                 );
     540            0 :             }
     541              : 
     542            0 :             CompleteDownload {
     543            0 :                 secondary_state,
     544            0 :                 completed_at: Instant::now(),
     545            0 :                 result
     546            0 :             }
     547            0 :         }.instrument(info_span!(parent: None, "secondary_download", tenant_id=%tenant_shard_id.tenant_id, shard_id=%tenant_shard_id.shard_slug()))))
     548            0 :     }
     549              : }
     550              : 
     551              : enum LayerAction {
     552              :     Download,
     553              :     NoAction,
     554              :     Skip,
     555              :     Touch,
     556              : }
     557              : 
     558              : /// This type is a convenience to group together the various functions involved in
     559              : /// freshening a secondary tenant.
     560              : struct TenantDownloader<'a> {
     561              :     conf: &'static PageServerConf,
     562              :     remote_storage: &'a GenericRemoteStorage,
     563              :     secondary_state: &'a SecondaryTenant,
     564              : }
     565              : 
     566              : /// Errors that may be encountered while updating a tenant
     567              : #[derive(thiserror::Error, Debug)]
     568              : enum UpdateError {
     569              :     /// This is not a true failure, but it's how a download indicates that it would like to be restarted by
     570              :     /// the scheduler, to pick up the latest heatmap
     571              :     #[error("Reached deadline, restarting downloads")]
     572              :     Restart,
     573              : 
     574              :     #[error("No remote data found")]
     575              :     NoData,
     576              :     #[error("Insufficient local storage space")]
     577              :     NoSpace,
     578              :     #[error("Failed to download")]
     579              :     DownloadError(DownloadError),
     580              :     #[error(transparent)]
     581              :     Deserialize(#[from] serde_json::Error),
     582              :     #[error("Cancelled")]
     583              :     Cancelled,
     584              :     #[error(transparent)]
     585              :     Other(#[from] anyhow::Error),
     586              : }
     587              : 
     588              : impl From<DownloadError> for UpdateError {
     589            0 :     fn from(value: DownloadError) -> Self {
     590            0 :         match &value {
     591            0 :             DownloadError::Cancelled => Self::Cancelled,
     592            0 :             DownloadError::NotFound => Self::NoData,
     593            0 :             _ => Self::DownloadError(value),
     594              :         }
     595            0 :     }
     596              : }
     597              : 
     598              : impl From<std::io::Error> for UpdateError {
     599            0 :     fn from(value: std::io::Error) -> Self {
     600            0 :         if let Some(nix::errno::Errno::ENOSPC) = value.raw_os_error().map(nix::errno::from_i32) {
     601            0 :             UpdateError::NoSpace
     602            0 :         } else if value
     603            0 :             .get_ref()
     604            0 :             .and_then(|x| x.downcast_ref::<DownloadError>())
     605            0 :             .is_some()
     606              :         {
     607            0 :             UpdateError::from(DownloadError::from(value))
     608              :         } else {
     609              :             // An I/O error from e.g. tokio::io::copy_buf is most likely a remote storage issue
     610            0 :             UpdateError::Other(anyhow::anyhow!(value))
     611              :         }
     612            0 :     }
     613              : }
     614              : 
     615              : impl<'a> TenantDownloader<'a> {
     616            0 :     fn new(
     617            0 :         conf: &'static PageServerConf,
     618            0 :         remote_storage: &'a GenericRemoteStorage,
     619            0 :         secondary_state: &'a SecondaryTenant,
     620            0 :     ) -> Self {
     621            0 :         Self {
     622            0 :             conf,
     623            0 :             remote_storage,
     624            0 :             secondary_state,
     625            0 :         }
     626            0 :     }
     627              : 
     628            0 :     async fn download(&self, ctx: &RequestContext) -> Result<(), UpdateError> {
     629            0 :         debug_assert_current_span_has_tenant_id();
     630              : 
     631              :         // For the duration of a download, we must hold the SecondaryTenant::gate, to ensure
     632              :         // cover our access to local storage.
     633            0 :         let Ok(_guard) = self.secondary_state.gate.enter() else {
     634              :             // Shutting down
     635            0 :             return Err(UpdateError::Cancelled);
     636              :         };
     637              : 
     638            0 :         let tenant_shard_id = self.secondary_state.get_tenant_shard_id();
     639            0 : 
     640            0 :         // We will use the etag from last successful download to make the download conditional on changes
     641            0 :         let last_download = self
     642            0 :             .secondary_state
     643            0 :             .detail
     644            0 :             .lock()
     645            0 :             .unwrap()
     646            0 :             .last_download
     647            0 :             .clone();
     648              : 
     649              :         // Download the tenant's heatmap
     650              :         let HeatMapModified {
     651            0 :             last_modified: heatmap_mtime,
     652            0 :             etag: heatmap_etag,
     653            0 :             bytes: heatmap_bytes,
     654            0 :         } = match tokio::select!(
     655            0 :             bytes = self.download_heatmap(last_download.as_ref().map(|d| &d.etag)) => {bytes?},
     656            0 :             _ = self.secondary_state.cancel.cancelled() => return Ok(())
     657              :         ) {
     658              :             HeatMapDownload::Unmodified => {
     659            0 :                 tracing::info!("Heatmap unchanged since last successful download");
     660            0 :                 return Ok(());
     661              :             }
     662            0 :             HeatMapDownload::Modified(m) => m,
     663            0 :         };
     664            0 : 
     665            0 :         // Heatmap storage location
     666            0 :         let heatmap_path = self.conf.tenant_heatmap_path(tenant_shard_id);
     667              : 
     668            0 :         let last_heatmap = if last_download.is_none() {
     669            0 :             match load_heatmap(&heatmap_path, ctx).await {
     670            0 :                 Ok(htm) => htm,
     671            0 :                 Err(e) => {
     672            0 :                     tracing::warn!("Couldn't load heatmap from {heatmap_path}: {e:?}");
     673            0 :                     None
     674              :                 }
     675              :             }
     676              :         } else {
     677            0 :             None
     678              :         };
     679              : 
     680            0 :         let last_heatmap_timelines = last_heatmap.as_ref().map(|htm| {
     681            0 :             htm.timelines
     682            0 :                 .iter()
     683            0 :                 .map(|tl| (tl.timeline_id, tl))
     684            0 :                 .collect::<HashMap<_, _>>()
     685            0 :         });
     686              : 
     687            0 :         let heatmap = serde_json::from_slice::<HeatMapTenant>(&heatmap_bytes)?;
     688              : 
     689            0 :         let temp_path = path_with_suffix_extension(&heatmap_path, TEMP_FILE_SUFFIX);
     690            0 :         let context_msg = format!("write tenant {tenant_shard_id} heatmap to {heatmap_path}");
     691            0 :         let heatmap_path_bg = heatmap_path.clone();
     692            0 :         VirtualFile::crashsafe_overwrite(heatmap_path_bg, temp_path, heatmap_bytes)
     693            0 :             .await
     694            0 :             .maybe_fatal_err(&context_msg)?;
     695              : 
     696            0 :         tracing::debug!(
     697            0 :             "Wrote local heatmap to {}, with {} timelines",
     698            0 :             heatmap_path,
     699            0 :             heatmap.timelines.len()
     700              :         );
     701              : 
     702              :         // Get or initialize the local disk state for the timelines we will update
     703            0 :         let mut timeline_states = HashMap::new();
     704            0 :         for timeline in &heatmap.timelines {
     705            0 :             let timeline_state = self
     706            0 :                 .secondary_state
     707            0 :                 .detail
     708            0 :                 .lock()
     709            0 :                 .unwrap()
     710            0 :                 .timelines
     711            0 :                 .get(&timeline.timeline_id)
     712            0 :                 .cloned();
     713              : 
     714            0 :             let timeline_state = match timeline_state {
     715            0 :                 Some(t) => t,
     716              :                 None => {
     717            0 :                     let last_heatmap =
     718            0 :                         last_heatmap_timelines
     719            0 :                             .as_ref()
     720            0 :                             .and_then(|last_heatmap_timelines| {
     721            0 :                                 last_heatmap_timelines.get(&timeline.timeline_id).copied()
     722            0 :                             });
     723              :                     // We have no existing state: need to scan local disk for layers first.
     724            0 :                     let timeline_state = init_timeline_state(
     725            0 :                         self.conf,
     726            0 :                         tenant_shard_id,
     727            0 :                         last_heatmap,
     728            0 :                         timeline,
     729            0 :                         &self.secondary_state.resident_size_metric,
     730            0 :                     )
     731            0 :                     .await;
     732              : 
     733              :                     // Re-acquire detail lock now that we're done with async load from local FS
     734            0 :                     self.secondary_state
     735            0 :                         .detail
     736            0 :                         .lock()
     737            0 :                         .unwrap()
     738            0 :                         .timelines
     739            0 :                         .insert(timeline.timeline_id, timeline_state.clone());
     740            0 :                     timeline_state
     741              :                 }
     742              :             };
     743              : 
     744            0 :             timeline_states.insert(timeline.timeline_id, timeline_state);
     745              :         }
     746              : 
     747              :         // Clean up any local layers that aren't in the heatmap.  We do this first for all timelines, on the general
     748              :         // principle that deletions should be done before writes wherever possible, and so that we can use this
     749              :         // phase to initialize our SecondaryProgress.
     750              :         {
     751            0 :             *self.secondary_state.progress.lock().unwrap() =
     752            0 :                 self.prepare_timelines(&heatmap, heatmap_mtime).await?;
     753              :         }
     754              : 
     755              :         // Calculate a deadline for downloads: if downloading takes longer than this, it is useful to drop out and start again,
     756              :         // so that we are always using reasonably a fresh heatmap.  Otherwise, if we had really huge content to download, we might
     757              :         // spend 10s of minutes downloading layers we don't need.
     758              :         // (see https://github.com/neondatabase/neon/issues/8182)
     759            0 :         let deadline = {
     760            0 :             let period = self
     761            0 :                 .secondary_state
     762            0 :                 .detail
     763            0 :                 .lock()
     764            0 :                 .unwrap()
     765            0 :                 .last_download
     766            0 :                 .as_ref()
     767            0 :                 .map(|d| d.upload_period)
     768            0 :                 .unwrap_or(DEFAULT_DOWNLOAD_INTERVAL);
     769            0 : 
     770            0 :             // Use double the period: we are not promising to complete within the period, this is just a heuristic
     771            0 :             // to keep using a "reasonably fresh" heatmap.
     772            0 :             Instant::now() + period * 2
     773              :         };
     774              : 
     775              :         // Download the layers in the heatmap
     776            0 :         for timeline in heatmap.timelines {
     777            0 :             let ctx = &ctx.with_scope_secondary_timeline(tenant_shard_id, &timeline.timeline_id);
     778            0 :             let timeline_state = timeline_states
     779            0 :                 .remove(&timeline.timeline_id)
     780            0 :                 .expect("Just populated above");
     781            0 : 
     782            0 :             if self.secondary_state.cancel.is_cancelled() {
     783            0 :                 tracing::debug!(
     784            0 :                     "Cancelled before downloading timeline {}",
     785              :                     timeline.timeline_id
     786              :                 );
     787            0 :                 return Ok(());
     788            0 :             }
     789            0 : 
     790            0 :             let timeline_id = timeline.timeline_id;
     791            0 :             self.download_timeline(timeline, timeline_state, deadline, ctx)
     792            0 :                 .instrument(tracing::info_span!(
     793              :                     "secondary_download_timeline",
     794              :                     tenant_id=%tenant_shard_id.tenant_id,
     795            0 :                     shard_id=%tenant_shard_id.shard_slug(),
     796              :                     %timeline_id
     797              :                 ))
     798            0 :                 .await?;
     799              :         }
     800              : 
     801              :         // Metrics consistency check in testing builds
     802            0 :         self.secondary_state.validate_metrics();
     803            0 :         // Only update last_etag after a full successful download: this way will not skip
     804            0 :         // the next download, even if the heatmap's actual etag is unchanged.
     805            0 :         self.secondary_state.detail.lock().unwrap().last_download = Some(DownloadSummary {
     806            0 :             etag: heatmap_etag,
     807            0 :             mtime: heatmap_mtime,
     808            0 :             upload_period: heatmap
     809            0 :                 .upload_period_ms
     810            0 :                 .map(|ms| Duration::from_millis(ms as u64))
     811            0 :                 .unwrap_or(DEFAULT_DOWNLOAD_INTERVAL),
     812            0 :         });
     813            0 : 
     814            0 :         // Robustness: we should have updated progress properly, but in case we didn't, make sure
     815            0 :         // we don't leave the tenant in a state where we claim to have successfully downloaded
     816            0 :         // everything, but our progress is incomplete.  The invariant here should be that if
     817            0 :         // we have set `last_download` to this heatmap's etag, then the next time we see that
     818            0 :         // etag we can safely do no work (i.e. we must be complete).
     819            0 :         let mut progress = self.secondary_state.progress.lock().unwrap();
     820            0 :         debug_assert!(progress.layers_downloaded == progress.layers_total);
     821            0 :         debug_assert!(progress.bytes_downloaded == progress.bytes_total);
     822            0 :         if progress.layers_downloaded != progress.layers_total
     823            0 :             || progress.bytes_downloaded != progress.bytes_total
     824              :         {
     825            0 :             tracing::warn!("Correcting drift in progress stats ({progress:?})");
     826            0 :             progress.layers_downloaded = progress.layers_total;
     827            0 :             progress.bytes_downloaded = progress.bytes_total;
     828            0 :         }
     829              : 
     830            0 :         Ok(())
     831            0 :     }
     832              : 
     833              :     /// Do any fast local cleanup that comes before the much slower process of downloading
     834              :     /// layers from remote storage.  In the process, initialize the SecondaryProgress object
     835              :     /// that will later be updated incrementally as we download layers.
     836            0 :     async fn prepare_timelines(
     837            0 :         &self,
     838            0 :         heatmap: &HeatMapTenant,
     839            0 :         heatmap_mtime: SystemTime,
     840            0 :     ) -> Result<SecondaryProgress, UpdateError> {
     841            0 :         let heatmap_stats = heatmap.get_stats();
     842            0 :         // We will construct a progress object, and then populate its initial "downloaded" numbers
     843            0 :         // while iterating through local layer state in [`Self::prepare_timelines`]
     844            0 :         let mut progress = SecondaryProgress {
     845            0 :             layers_total: heatmap_stats.layers,
     846            0 :             bytes_total: heatmap_stats.bytes,
     847            0 :             heatmap_mtime: Some(serde_system_time::SystemTime(heatmap_mtime)),
     848            0 :             layers_downloaded: 0,
     849            0 :             bytes_downloaded: 0,
     850            0 :         };
     851            0 : 
     852            0 :         // Also expose heatmap bytes_total as a metric
     853            0 :         self.secondary_state
     854            0 :             .heatmap_total_size_metric
     855            0 :             .set(heatmap_stats.bytes);
     856            0 : 
     857            0 :         // Accumulate list of things to delete while holding the detail lock, for execution after dropping the lock
     858            0 :         let mut delete_layers = Vec::new();
     859            0 :         let mut delete_timelines = Vec::new();
     860            0 :         {
     861            0 :             let mut detail = self.secondary_state.detail.lock().unwrap();
     862            0 :             for (timeline_id, timeline_state) in &mut detail.timelines {
     863            0 :                 let Some(heatmap_timeline_index) = heatmap
     864            0 :                     .timelines
     865            0 :                     .iter()
     866            0 :                     .position(|t| t.timeline_id == *timeline_id)
     867              :                 else {
     868              :                     // This timeline is no longer referenced in the heatmap: delete it locally
     869            0 :                     delete_timelines.push(*timeline_id);
     870            0 :                     continue;
     871              :                 };
     872              : 
     873            0 :                 let heatmap_timeline = heatmap.timelines.get(heatmap_timeline_index).unwrap();
     874            0 : 
     875            0 :                 let layers_in_heatmap = heatmap_timeline
     876            0 :                     .hot_layers()
     877            0 :                     .map(|l| (&l.name, l.metadata.generation))
     878            0 :                     .collect::<HashSet<_>>();
     879            0 :                 let layers_on_disk = timeline_state
     880            0 :                     .on_disk_layers
     881            0 :                     .iter()
     882            0 :                     .map(|l| (l.0, l.1.metadata.generation))
     883            0 :                     .collect::<HashSet<_>>();
     884            0 : 
     885            0 :                 let mut layer_count = layers_on_disk.len();
     886            0 :                 let mut layer_byte_count: u64 = timeline_state
     887            0 :                     .on_disk_layers
     888            0 :                     .values()
     889            0 :                     .map(|l| l.metadata.file_size)
     890            0 :                     .sum();
     891              : 
     892              :                 // Remove on-disk layers that are no longer present in heatmap
     893            0 :                 for (layer_file_name, generation) in layers_on_disk.difference(&layers_in_heatmap) {
     894            0 :                     layer_count -= 1;
     895            0 :                     layer_byte_count -= timeline_state
     896            0 :                         .on_disk_layers
     897            0 :                         .get(layer_file_name)
     898            0 :                         .unwrap()
     899            0 :                         .metadata
     900            0 :                         .file_size;
     901            0 : 
     902            0 :                     let local_path = local_layer_path(
     903            0 :                         self.conf,
     904            0 :                         self.secondary_state.get_tenant_shard_id(),
     905            0 :                         timeline_id,
     906            0 :                         layer_file_name,
     907            0 :                         generation,
     908            0 :                     );
     909            0 : 
     910            0 :                     delete_layers.push((*timeline_id, (*layer_file_name).clone(), local_path));
     911            0 :                 }
     912              : 
     913            0 :                 progress.bytes_downloaded += layer_byte_count;
     914            0 :                 progress.layers_downloaded += layer_count;
     915              :             }
     916              : 
     917            0 :             for delete_timeline in &delete_timelines {
     918            0 :                 // We haven't removed from disk yet, but optimistically remove from in-memory state: if removal
     919            0 :                 // from disk fails that will be a fatal error.
     920            0 :                 detail.remove_timeline(delete_timeline, &self.secondary_state.resident_size_metric);
     921            0 :             }
     922              :         }
     923              : 
     924              :         // Execute accumulated deletions
     925            0 :         for (timeline_id, layer_name, local_path) in delete_layers {
     926            0 :             tracing::info!(timeline_id=%timeline_id, "Removing secondary local layer {layer_name} because it's absent in heatmap",);
     927              : 
     928            0 :             tokio::fs::remove_file(&local_path)
     929            0 :                 .await
     930            0 :                 .or_else(fs_ext::ignore_not_found)
     931            0 :                 .maybe_fatal_err("Removing secondary layer")?;
     932              : 
     933              :             // Update in-memory housekeeping to reflect the absence of the deleted layer
     934            0 :             let mut detail = self.secondary_state.detail.lock().unwrap();
     935            0 :             let Some(timeline_state) = detail.timelines.get_mut(&timeline_id) else {
     936            0 :                 continue;
     937              :             };
     938            0 :             timeline_state.remove_layer(&layer_name, &self.secondary_state.resident_size_metric);
     939              :         }
     940              : 
     941            0 :         for timeline_id in delete_timelines {
     942            0 :             let timeline_path = self
     943            0 :                 .conf
     944            0 :                 .timeline_path(self.secondary_state.get_tenant_shard_id(), &timeline_id);
     945            0 :             tracing::info!(timeline_id=%timeline_id,
     946            0 :                 "Timeline no longer in heatmap, removing from secondary location"
     947              :             );
     948            0 :             tokio::fs::remove_dir_all(&timeline_path)
     949            0 :                 .await
     950            0 :                 .or_else(fs_ext::ignore_not_found)
     951            0 :                 .maybe_fatal_err("Removing secondary timeline")?;
     952              :         }
     953              : 
     954            0 :         Ok(progress)
     955            0 :     }
     956              : 
     957              :     /// Returns downloaded bytes if the etag differs from `prev_etag`, or None if the object
     958              :     /// still matches `prev_etag`.
     959            0 :     async fn download_heatmap(
     960            0 :         &self,
     961            0 :         prev_etag: Option<&Etag>,
     962            0 :     ) -> Result<HeatMapDownload, UpdateError> {
     963            0 :         debug_assert_current_span_has_tenant_id();
     964            0 :         let tenant_shard_id = self.secondary_state.get_tenant_shard_id();
     965            0 :         tracing::debug!("Downloading heatmap for secondary tenant",);
     966              : 
     967            0 :         let heatmap_path = remote_heatmap_path(tenant_shard_id);
     968            0 :         let cancel = &self.secondary_state.cancel;
     969            0 :         let opts = DownloadOpts {
     970            0 :             etag: prev_etag.cloned(),
     971            0 :             kind: DownloadKind::Small,
     972            0 :             ..Default::default()
     973            0 :         };
     974            0 : 
     975            0 :         backoff::retry(
     976            0 :             || async {
     977            0 :                 let download = match self
     978            0 :                     .remote_storage
     979            0 :                     .download(&heatmap_path, &opts, cancel)
     980            0 :                     .await
     981              :                 {
     982            0 :                     Ok(download) => download,
     983            0 :                     Err(DownloadError::Unmodified) => return Ok(HeatMapDownload::Unmodified),
     984            0 :                     Err(err) => return Err(err.into()),
     985              :                 };
     986              : 
     987            0 :                 let mut heatmap_bytes = Vec::new();
     988            0 :                 let mut body = tokio_util::io::StreamReader::new(download.download_stream);
     989            0 :                 let _size = tokio::io::copy_buf(&mut body, &mut heatmap_bytes).await?;
     990            0 :                 Ok(HeatMapDownload::Modified(HeatMapModified {
     991            0 :                     etag: download.etag,
     992            0 :                     last_modified: download.last_modified,
     993            0 :                     bytes: heatmap_bytes,
     994            0 :                 }))
     995            0 :             },
     996            0 :             |e| matches!(e, UpdateError::NoData | UpdateError::Cancelled),
     997            0 :             FAILED_DOWNLOAD_WARN_THRESHOLD,
     998            0 :             FAILED_REMOTE_OP_RETRIES,
     999            0 :             "download heatmap",
    1000            0 :             cancel,
    1001            0 :         )
    1002            0 :         .await
    1003            0 :         .ok_or_else(|| UpdateError::Cancelled)
    1004            0 :         .and_then(|x| x)
    1005            0 :         .inspect(|_| SECONDARY_MODE.download_heatmap.inc())
    1006            0 :     }
    1007              : 
    1008              :     /// Download heatmap layers that are not present on local disk, or update their
    1009              :     /// access time if they are already present.
    1010            0 :     async fn download_timeline_layers(
    1011            0 :         &self,
    1012            0 :         tenant_shard_id: &TenantShardId,
    1013            0 :         timeline: HeatMapTimeline,
    1014            0 :         timeline_state: SecondaryDetailTimeline,
    1015            0 :         deadline: Instant,
    1016            0 :         ctx: &RequestContext,
    1017            0 :     ) -> (Result<(), UpdateError>, Vec<HeatMapLayer>) {
    1018            0 :         // Accumulate updates to the state
    1019            0 :         let mut touched = Vec::new();
    1020            0 : 
    1021            0 :         let timeline_id = timeline.timeline_id;
    1022            0 :         for layer in timeline.into_hot_layers() {
    1023            0 :             if self.secondary_state.cancel.is_cancelled() {
    1024            0 :                 tracing::debug!("Cancelled -- dropping out of layer loop");
    1025            0 :                 return (Err(UpdateError::Cancelled), touched);
    1026            0 :             }
    1027            0 : 
    1028            0 :             if Instant::now() > deadline {
    1029              :                 // We've been running downloads for a while, restart to download latest heatmap.
    1030            0 :                 return (Err(UpdateError::Restart), touched);
    1031            0 :             }
    1032            0 : 
    1033            0 :             match self.layer_action(&timeline_state, &layer).await {
    1034            0 :                 LayerAction::Download => (),
    1035            0 :                 LayerAction::NoAction => continue,
    1036              :                 LayerAction::Skip => {
    1037            0 :                     self.skip_layer(layer);
    1038            0 :                     continue;
    1039              :                 }
    1040              :                 LayerAction::Touch => {
    1041            0 :                     touched.push(layer);
    1042            0 :                     continue;
    1043              :                 }
    1044              :             }
    1045              : 
    1046            0 :             match self
    1047            0 :                 .download_layer(tenant_shard_id, &timeline_id, layer, ctx)
    1048            0 :                 .await
    1049              :             {
    1050            0 :                 Ok(Some(layer)) => touched.push(layer),
    1051            0 :                 Ok(None) => {
    1052            0 :                     // Not an error but we didn't download it: remote layer is missing.  Don't add it to the list of
    1053            0 :                     // things to consider touched.
    1054            0 :                 }
    1055            0 :                 Err(e) => {
    1056            0 :                     return (Err(e), touched);
    1057              :                 }
    1058              :             }
    1059              :         }
    1060              : 
    1061            0 :         (Ok(()), touched)
    1062            0 :     }
    1063              : 
    1064            0 :     async fn layer_action(
    1065            0 :         &self,
    1066            0 :         timeline_state: &SecondaryDetailTimeline,
    1067            0 :         layer: &HeatMapLayer,
    1068            0 :     ) -> LayerAction {
    1069              :         // Existing on-disk layers: just update their access time.
    1070            0 :         if let Some(on_disk) = timeline_state.on_disk_layers.get(&layer.name) {
    1071            0 :             tracing::debug!("Layer {} is already on disk", layer.name);
    1072              : 
    1073            0 :             if cfg!(debug_assertions) {
    1074              :                 // Debug for https://github.com/neondatabase/neon/issues/6966: check that the files we think
    1075              :                 // are already present on disk are really there.
    1076            0 :                 match tokio::fs::metadata(&on_disk.local_path).await {
    1077            0 :                     Ok(meta) => {
    1078            0 :                         tracing::debug!(
    1079            0 :                             "Layer {} present at {}, size {}",
    1080            0 :                             layer.name,
    1081            0 :                             on_disk.local_path,
    1082            0 :                             meta.len(),
    1083              :                         );
    1084              :                     }
    1085            0 :                     Err(e) => {
    1086            0 :                         tracing::warn!(
    1087            0 :                             "Layer {} not found at {} ({})",
    1088              :                             layer.name,
    1089              :                             on_disk.local_path,
    1090              :                             e
    1091              :                         );
    1092            0 :                         debug_assert!(false);
    1093              :                     }
    1094              :                 }
    1095            0 :             }
    1096              : 
    1097            0 :             if on_disk.metadata.generation_file_size() != layer.metadata.generation_file_size() {
    1098            0 :                 tracing::info!(
    1099            0 :                     "Re-downloading layer {} with changed size or generation: {:?}->{:?}",
    1100            0 :                     layer.name,
    1101            0 :                     on_disk.metadata.generation_file_size(),
    1102            0 :                     layer.metadata.generation_file_size()
    1103              :                 );
    1104            0 :                 return LayerAction::Download;
    1105            0 :             }
    1106            0 :             if on_disk.metadata != layer.metadata || on_disk.access_time != layer.access_time {
    1107              :                 // We already have this layer on disk.  Update its access time.
    1108            0 :                 tracing::debug!(
    1109            0 :                     "Access time updated for layer {}: {} -> {}",
    1110            0 :                     layer.name,
    1111            0 :                     strftime(&on_disk.access_time),
    1112            0 :                     strftime(&layer.access_time)
    1113              :                 );
    1114            0 :                 return LayerAction::Touch;
    1115            0 :             }
    1116            0 :             return LayerAction::NoAction;
    1117              :         } else {
    1118            0 :             tracing::debug!("Layer {} not present on disk yet", layer.name);
    1119              :         }
    1120              : 
    1121              :         // Eviction: if we evicted a layer, then do not re-download it unless it was accessed more
    1122              :         // recently than it was evicted.
    1123            0 :         if let Some(evicted_at) = timeline_state.evicted_at.get(&layer.name) {
    1124            0 :             if &layer.access_time > evicted_at {
    1125            0 :                 tracing::info!(
    1126            0 :                     "Re-downloading evicted layer {}, accessed at {}, evicted at {}",
    1127            0 :                     layer.name,
    1128            0 :                     strftime(&layer.access_time),
    1129            0 :                     strftime(evicted_at)
    1130              :                 );
    1131              :             } else {
    1132            0 :                 tracing::trace!(
    1133            0 :                     "Not re-downloading evicted layer {}, accessed at {}, evicted at {}",
    1134            0 :                     layer.name,
    1135            0 :                     strftime(&layer.access_time),
    1136            0 :                     strftime(evicted_at)
    1137              :                 );
    1138            0 :                 return LayerAction::Skip;
    1139              :             }
    1140            0 :         }
    1141            0 :         LayerAction::Download
    1142            0 :     }
    1143              : 
    1144            0 :     async fn download_timeline(
    1145            0 :         &self,
    1146            0 :         timeline: HeatMapTimeline,
    1147            0 :         timeline_state: SecondaryDetailTimeline,
    1148            0 :         deadline: Instant,
    1149            0 :         ctx: &RequestContext,
    1150            0 :     ) -> Result<(), UpdateError> {
    1151            0 :         debug_assert_current_span_has_tenant_and_timeline_id();
    1152            0 :         let tenant_shard_id = self.secondary_state.get_tenant_shard_id();
    1153            0 :         let timeline_id = timeline.timeline_id;
    1154            0 : 
    1155            0 :         tracing::debug!(timeline_id=%timeline_id, "Downloading layers, {} in heatmap", timeline.hot_layers().count());
    1156              : 
    1157            0 :         let (result, touched) = self
    1158            0 :             .download_timeline_layers(tenant_shard_id, timeline, timeline_state, deadline, ctx)
    1159            0 :             .await;
    1160              : 
    1161              :         // Write updates to state to record layers we just downloaded or touched, irrespective of whether the overall result was successful
    1162              :         {
    1163            0 :             let mut detail = self.secondary_state.detail.lock().unwrap();
    1164            0 :             let timeline_detail = detail.timelines.entry(timeline_id).or_default();
    1165            0 : 
    1166            0 :             tracing::info!("Wrote timeline_detail for {} touched layers", touched.len());
    1167            0 :             touched.into_iter().for_each(|t| {
    1168            0 :                 timeline_detail.touch_layer(
    1169            0 :                     self.conf,
    1170            0 :                     tenant_shard_id,
    1171            0 :                     &timeline_id,
    1172            0 :                     &t,
    1173            0 :                     &self.secondary_state.resident_size_metric,
    1174            0 :                     || {
    1175            0 :                         local_layer_path(
    1176            0 :                             self.conf,
    1177            0 :                             tenant_shard_id,
    1178            0 :                             &timeline_id,
    1179            0 :                             &t.name,
    1180            0 :                             &t.metadata.generation,
    1181            0 :                         )
    1182            0 :                     },
    1183            0 :                 )
    1184            0 :             });
    1185            0 :         }
    1186            0 : 
    1187            0 :         result
    1188            0 :     }
    1189              : 
    1190              :     /// Call this during timeline download if a layer will _not_ be downloaded, to update progress statistics
    1191            0 :     fn skip_layer(&self, layer: HeatMapLayer) {
    1192            0 :         let mut progress = self.secondary_state.progress.lock().unwrap();
    1193            0 :         progress.layers_total = progress.layers_total.saturating_sub(1);
    1194            0 :         progress.bytes_total = progress
    1195            0 :             .bytes_total
    1196            0 :             .saturating_sub(layer.metadata.file_size);
    1197            0 :     }
    1198              : 
    1199            0 :     async fn download_layer(
    1200            0 :         &self,
    1201            0 :         tenant_shard_id: &TenantShardId,
    1202            0 :         timeline_id: &TimelineId,
    1203            0 :         layer: HeatMapLayer,
    1204            0 :         ctx: &RequestContext,
    1205            0 :     ) -> Result<Option<HeatMapLayer>, UpdateError> {
    1206            0 :         // Failpoints for simulating slow remote storage
    1207            0 :         failpoint_support::sleep_millis_async!(
    1208              :             "secondary-layer-download-sleep",
    1209            0 :             &self.secondary_state.cancel
    1210              :         );
    1211              : 
    1212            0 :         pausable_failpoint!("secondary-layer-download-pausable");
    1213              : 
    1214            0 :         let local_path = local_layer_path(
    1215            0 :             self.conf,
    1216            0 :             tenant_shard_id,
    1217            0 :             timeline_id,
    1218            0 :             &layer.name,
    1219            0 :             &layer.metadata.generation,
    1220            0 :         );
    1221            0 : 
    1222            0 :         // Note: no backoff::retry wrapper here because download_layer_file does its own retries internally
    1223            0 :         tracing::info!(
    1224            0 :             "Starting download of layer {}, size {}",
    1225              :             layer.name,
    1226              :             layer.metadata.file_size
    1227              :         );
    1228            0 :         let downloaded_bytes = download_layer_file(
    1229            0 :             self.conf,
    1230            0 :             self.remote_storage,
    1231            0 :             *tenant_shard_id,
    1232            0 :             *timeline_id,
    1233            0 :             &layer.name,
    1234            0 :             &layer.metadata,
    1235            0 :             &local_path,
    1236            0 :             &self.secondary_state.gate,
    1237            0 :             &self.secondary_state.cancel,
    1238            0 :             ctx,
    1239            0 :         )
    1240            0 :         .await;
    1241              : 
    1242            0 :         let downloaded_bytes = match downloaded_bytes {
    1243            0 :             Ok(bytes) => bytes,
    1244              :             Err(DownloadError::NotFound) => {
    1245              :                 // A heatmap might be out of date and refer to a layer that doesn't exist any more.
    1246              :                 // This is harmless: continue to download the next layer. It is expected during compaction
    1247              :                 // GC.
    1248            0 :                 tracing::debug!(
    1249            0 :                     "Skipped downloading missing layer {}, raced with compaction/gc?",
    1250              :                     layer.name
    1251              :                 );
    1252            0 :                 self.skip_layer(layer);
    1253            0 : 
    1254            0 :                 return Ok(None);
    1255              :             }
    1256            0 :             Err(e) => return Err(e.into()),
    1257              :         };
    1258              : 
    1259            0 :         if downloaded_bytes != layer.metadata.file_size {
    1260            0 :             let local_path = local_layer_path(
    1261            0 :                 self.conf,
    1262            0 :                 tenant_shard_id,
    1263            0 :                 timeline_id,
    1264            0 :                 &layer.name,
    1265            0 :                 &layer.metadata.generation,
    1266            0 :             );
    1267            0 : 
    1268            0 :             tracing::warn!(
    1269            0 :                 "Downloaded layer {} with unexpected size {} != {}.  Removing download.",
    1270              :                 layer.name,
    1271              :                 downloaded_bytes,
    1272              :                 layer.metadata.file_size
    1273              :             );
    1274              : 
    1275            0 :             tokio::fs::remove_file(&local_path)
    1276            0 :                 .await
    1277            0 :                 .or_else(fs_ext::ignore_not_found)?;
    1278              :         } else {
    1279            0 :             tracing::info!("Downloaded layer {}, size {}", layer.name, downloaded_bytes);
    1280            0 :             let mut progress = self.secondary_state.progress.lock().unwrap();
    1281            0 :             progress.bytes_downloaded += downloaded_bytes;
    1282            0 :             progress.layers_downloaded += 1;
    1283              :         }
    1284              : 
    1285            0 :         SECONDARY_MODE.download_layer.inc();
    1286            0 : 
    1287            0 :         Ok(Some(layer))
    1288            0 :     }
    1289              : }
    1290              : 
    1291              : /// Scan local storage and build up Layer objects based on the metadata in a HeatMapTimeline
    1292            0 : async fn init_timeline_state(
    1293            0 :     conf: &'static PageServerConf,
    1294            0 :     tenant_shard_id: &TenantShardId,
    1295            0 :     last_heatmap: Option<&HeatMapTimeline>,
    1296            0 :     heatmap: &HeatMapTimeline,
    1297            0 :     resident_metric: &UIntGauge,
    1298            0 : ) -> SecondaryDetailTimeline {
    1299            0 :     let timeline_path = conf.timeline_path(tenant_shard_id, &heatmap.timeline_id);
    1300            0 :     let mut detail = SecondaryDetailTimeline::default();
    1301              : 
    1302            0 :     let mut dir = match tokio::fs::read_dir(&timeline_path).await {
    1303            0 :         Ok(d) => d,
    1304            0 :         Err(e) => {
    1305            0 :             if e.kind() == std::io::ErrorKind::NotFound {
    1306            0 :                 let context = format!("Creating timeline directory {timeline_path}");
    1307            0 :                 tracing::info!("{}", context);
    1308            0 :                 tokio::fs::create_dir_all(&timeline_path)
    1309            0 :                     .await
    1310            0 :                     .fatal_err(&context);
    1311            0 : 
    1312            0 :                 // No entries to report: drop out.
    1313            0 :                 return detail;
    1314              :             } else {
    1315            0 :                 on_fatal_io_error(&e, &format!("Reading timeline dir {timeline_path}"));
    1316              :             }
    1317              :         }
    1318              :     };
    1319              : 
    1320              :     // As we iterate through layers found on disk, we will look up their metadata from this map.
    1321              :     // Layers not present in metadata will be discarded.
    1322            0 :     let heatmap_metadata: HashMap<&LayerName, &HeatMapLayer> =
    1323            0 :         heatmap.hot_layers().map(|l| (&l.name, l)).collect();
    1324              : 
    1325            0 :     let last_heatmap_metadata: HashMap<&LayerName, &HeatMapLayer> =
    1326            0 :         if let Some(last_heatmap) = last_heatmap {
    1327            0 :             last_heatmap.hot_layers().map(|l| (&l.name, l)).collect()
    1328              :         } else {
    1329            0 :             HashMap::new()
    1330              :         };
    1331              : 
    1332            0 :     while let Some(dentry) = dir
    1333            0 :         .next_entry()
    1334            0 :         .await
    1335            0 :         .fatal_err(&format!("Listing {timeline_path}"))
    1336              :     {
    1337            0 :         let Ok(file_path) = Utf8PathBuf::from_path_buf(dentry.path()) else {
    1338            0 :             tracing::warn!("Malformed filename at {}", dentry.path().to_string_lossy());
    1339            0 :             continue;
    1340              :         };
    1341            0 :         let local_meta = dentry
    1342            0 :             .metadata()
    1343            0 :             .await
    1344            0 :             .fatal_err(&format!("Read metadata on {}", file_path));
    1345            0 : 
    1346            0 :         let file_name = file_path.file_name().expect("created it from the dentry");
    1347            0 :         if crate::is_temporary(&file_path)
    1348            0 :             || is_temp_download_file(&file_path)
    1349            0 :             || is_ephemeral_file(file_name)
    1350              :         {
    1351              :             // Temporary files are frequently left behind from restarting during downloads
    1352            0 :             tracing::info!("Cleaning up temporary file {file_path}");
    1353            0 :             if let Err(e) = tokio::fs::remove_file(&file_path)
    1354            0 :                 .await
    1355            0 :                 .or_else(fs_ext::ignore_not_found)
    1356              :             {
    1357            0 :                 tracing::error!("Failed to remove temporary file {file_path}: {e}");
    1358            0 :             }
    1359            0 :             continue;
    1360            0 :         }
    1361            0 : 
    1362            0 :         match LayerName::from_str(file_name) {
    1363            0 :             Ok(name) => {
    1364            0 :                 let remote_meta = heatmap_metadata.get(&name);
    1365            0 :                 let last_meta = last_heatmap_metadata.get(&name);
    1366            0 :                 let mut remove = false;
    1367            0 :                 match remote_meta {
    1368            0 :                     Some(remote_meta) => {
    1369            0 :                         let last_meta_generation_file_size = last_meta
    1370            0 :                             .map(|m| m.metadata.generation_file_size())
    1371            0 :                             .unwrap_or(remote_meta.metadata.generation_file_size());
    1372            0 :                         // TODO: checksums for layers (https://github.com/neondatabase/neon/issues/2784)
    1373            0 :                         if remote_meta.metadata.generation_file_size()
    1374            0 :                             != last_meta_generation_file_size
    1375              :                         {
    1376            0 :                             tracing::info!(
    1377            0 :                                 "Removing local layer {name} as on-disk json metadata has different generation or file size from remote: {:?} -> {:?}",
    1378            0 :                                 last_meta_generation_file_size,
    1379            0 :                                 remote_meta.metadata.generation_file_size()
    1380              :                             );
    1381            0 :                             remove = true;
    1382            0 :                         } else if local_meta.len() != remote_meta.metadata.file_size {
    1383              :                             // This can happen in the presence of race conditions: the remote and on-disk metadata have changed, but we haven't had
    1384              :                             // the chance yet to download the new layer to disk, before the process restarted.
    1385            0 :                             tracing::info!(
    1386            0 :                                 "Removing local layer {name} with unexpected local size {} != {}",
    1387            0 :                                 local_meta.len(),
    1388              :                                 remote_meta.metadata.file_size
    1389              :                             );
    1390            0 :                             remove = true;
    1391            0 :                         } else {
    1392            0 :                             // We expect the access time to be initialized immediately afterwards, when
    1393            0 :                             // the latest heatmap is applied to the state.
    1394            0 :                             detail.touch_layer(
    1395            0 :                                 conf,
    1396            0 :                                 tenant_shard_id,
    1397            0 :                                 &heatmap.timeline_id,
    1398            0 :                                 remote_meta,
    1399            0 :                                 resident_metric,
    1400            0 :                                 || file_path,
    1401            0 :                             );
    1402            0 :                         }
    1403              :                     }
    1404              :                     None => {
    1405              :                         // FIXME: consider some optimization when transitioning from attached to secondary: maybe
    1406              :                         // wait until we have seen a heatmap that is more recent than the most recent on-disk state?  Otherwise
    1407              :                         // we will end up deleting any layers which were created+uploaded more recently than the heatmap.
    1408            0 :                         tracing::info!(
    1409            0 :                             "Removing secondary local layer {} because it's absent in heatmap",
    1410              :                             name
    1411              :                         );
    1412            0 :                         remove = true;
    1413              :                     }
    1414              :                 }
    1415            0 :                 if remove {
    1416            0 :                     tokio::fs::remove_file(&dentry.path())
    1417            0 :                         .await
    1418            0 :                         .or_else(fs_ext::ignore_not_found)
    1419            0 :                         .fatal_err(&format!(
    1420            0 :                             "Removing layer {}",
    1421            0 :                             dentry.path().to_string_lossy()
    1422            0 :                         ));
    1423            0 :                 }
    1424              :             }
    1425              :             Err(_) => {
    1426              :                 // Ignore it.
    1427            0 :                 tracing::warn!("Unexpected file in timeline directory: {file_name}");
    1428              :             }
    1429              :         }
    1430              :     }
    1431              : 
    1432            0 :     detail
    1433            0 : }
    1434              : 
    1435              : /// Loads a json-encoded heatmap file from the provided on-disk path
    1436            0 : async fn load_heatmap(
    1437            0 :     path: &Utf8PathBuf,
    1438            0 :     ctx: &RequestContext,
    1439            0 : ) -> Result<Option<HeatMapTenant>, anyhow::Error> {
    1440            0 :     let mut file = match VirtualFile::open(path, ctx).await {
    1441            0 :         Ok(file) => file,
    1442            0 :         Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
    1443            0 :         Err(e) => Err(e)?,
    1444              :     };
    1445            0 :     let st = file.read_to_string(ctx).await?;
    1446            0 :     let htm = serde_json::from_str(&st)?;
    1447            0 :     Ok(Some(htm))
    1448            0 : }
        

Generated by: LCOV version 2.1-beta