LCOV - code coverage report
Current view: top level - pageserver/src/tenant - remote_timeline_client.rs (source / functions) Coverage Total Hit
Test: 4f58e98c51285c7fa348e0b410c88a10caf68ad2.info Lines: 65.5 % 2018 1322
Test Date: 2025-01-07 20:58:07 Functions: 53.5 % 187 100

            Line data    Source code
       1              : //! This module manages synchronizing local FS with remote storage.
       2              : //!
       3              : //! # Overview
       4              : //!
       5              : //! * [`RemoteTimelineClient`] provides functions related to upload/download of a particular timeline.
       6              : //!   It contains a queue of pending uploads, and manages the queue, performing uploads in parallel
       7              : //!   when it's safe to do so.
       8              : //!
       9              : //! * Stand-alone function, [`list_remote_timelines`], to get list of timelines of a tenant.
      10              : //!
      11              : //! These functions use the low-level remote storage client, [`remote_storage::RemoteStorage`].
      12              : //!
      13              : //! # APIs & How To Use Them
      14              : //!
      15              : //! There is a [RemoteTimelineClient] for each [Timeline][`crate::tenant::Timeline`] in the system,
      16              : //! unless the pageserver is configured without remote storage.
      17              : //!
      18              : //! We allocate the client instance in [Timeline][`crate::tenant::Timeline`], i.e.,
      19              : //! either in [`crate::tenant::mgr`] during startup or when creating a new
      20              : //! timeline.
      21              : //! However, the client does not become ready for use until we've initialized its upload queue:
      22              : //!
      23              : //! - For timelines that already have some state on the remote storage, we use
      24              : //!   [`RemoteTimelineClient::init_upload_queue`] .
      25              : //! - For newly created timelines, we use
      26              : //!   [`RemoteTimelineClient::init_upload_queue_for_empty_remote`].
      27              : //!
      28              : //! The former takes the remote's [`IndexPart`] as an argument, possibly retrieved
      29              : //! using [`list_remote_timelines`]. We'll elaborate on [`IndexPart`] in the next section.
      30              : //!
      31              : //! Whenever we've created/updated/deleted a file in a timeline directory, we schedule
      32              : //! the corresponding remote operation with the timeline's [`RemoteTimelineClient`]:
      33              : //!
      34              : //! - [`RemoteTimelineClient::schedule_layer_file_upload`]  when we've created a new layer file.
      35              : //! - [`RemoteTimelineClient::schedule_index_upload_for_metadata_update`] when we've updated the timeline metadata file.
      36              : //! - [`RemoteTimelineClient::schedule_index_upload_for_file_changes`] to upload an updated index file, after we've scheduled file uploads
      37              : //! - [`RemoteTimelineClient::schedule_layer_file_deletion`] when we've deleted one or more layer files.
      38              : //!
      39              : //! Internally, these functions create [`UploadOp`]s and put them in a queue.
      40              : //!
      41              : //! There are also APIs for downloading files.
      42              : //! These are not part of the aforementioned queuing and will not be discussed
      43              : //! further here, except in the section covering tenant attach.
      44              : //!
      45              : //! # Remote Storage Structure & [`IndexPart`] Index File
      46              : //!
      47              : //! The "directory structure" in the remote storage mirrors the local directory structure, with paths
      48              : //! like `tenants/<tenant_id>/timelines/<timeline_id>/<layer filename>`.
      49              : //! Yet instead of keeping the `metadata` file remotely, we wrap it with more
      50              : //! data in an "index file" aka [`IndexPart`], containing the list of **all** remote
      51              : //! files for a given timeline.
      52              : //! If a file is not referenced from [`IndexPart`], it's not part of the remote storage state.
      53              : //!
      54              : //! Having the `IndexPart` also avoids expensive and slow `S3 list` commands.
      55              : //!
      56              : //! # Consistency
      57              : //!
      58              : //! To have a consistent remote structure, it's important that uploads and
      59              : //! deletions are performed in the right order. For example, the index file
      60              : //! contains a list of layer files, so it must not be uploaded until all the
      61              : //! layer files that are in its list have been successfully uploaded.
      62              : //!
      63              : //! The contract between client and its user is that the user is responsible of
      64              : //! scheduling operations in an order that keeps the remote consistent as
      65              : //! described above.
      66              : //! From the user's perspective, the operations are executed sequentially.
      67              : //! Internally, the client knows which operations can be performed in parallel,
      68              : //! and which operations act like a "barrier" that require preceding operations
      69              : //! to finish. The calling code just needs to call the schedule-functions in the
      70              : //! correct order, and the client will parallelize the operations in a way that
      71              : //! is safe.
      72              : //!
      73              : //! The caller should be careful with deletion, though. They should not delete
      74              : //! local files that have been scheduled for upload but not yet finished uploading.
      75              : //! Otherwise the upload will fail. To wait for an upload to finish, use
      76              : //! the 'wait_completion' function (more on that later.)
      77              : //!
      78              : //! All of this relies on the following invariants:
      79              : //!
      80              : //! - We rely on read-after write consistency in the remote storage.
      81              : //! - Layer files are immutable
      82              : //!
      83              : //! NB: Pageserver assumes that it has exclusive write access to the tenant in remote
      84              : //! storage. Different tenants can be attached to different pageservers, but if the
      85              : //! same tenant is attached to two pageservers at the same time, they will overwrite
      86              : //! each other's index file updates, and confusion will ensue. There's no interlock or
      87              : //! mechanism to detect that in the pageserver, we rely on the control plane to ensure
      88              : //! that that doesn't happen.
      89              : //!
      90              : //! ## Implementation Note
      91              : //!
      92              : //! The *actual* remote state lags behind the *desired* remote state while
      93              : //! there are in-flight operations.
      94              : //! We keep track of the desired remote state in [`UploadQueueInitialized::dirty`].
      95              : //! It is initialized based on the [`IndexPart`] that was passed during init
      96              : //! and updated with every `schedule_*` function call.
      97              : //! All this is necessary necessary to compute the future [`IndexPart`]s
      98              : //! when scheduling an operation while other operations that also affect the
      99              : //! remote [`IndexPart`] are in flight.
     100              : //!
     101              : //! # Retries & Error Handling
     102              : //!
     103              : //! The client retries operations indefinitely, using exponential back-off.
     104              : //! There is no way to force a retry, i.e., interrupt the back-off.
     105              : //! This could be built easily.
     106              : //!
     107              : //! # Cancellation
     108              : //!
     109              : //! The operations execute as plain [`task_mgr`] tasks, scoped to
     110              : //! the client's tenant and timeline.
     111              : //! Dropping the client will drop queued operations but not executing operations.
     112              : //! These will complete unless the `task_mgr` tasks are cancelled using `task_mgr`
     113              : //! APIs, e.g., during pageserver shutdown, timeline delete, or tenant detach.
     114              : //!
     115              : //! # Completion
     116              : //!
     117              : //! Once an operation has completed, we update [`UploadQueueInitialized::clean`] immediately,
     118              : //! and submit a request through the DeletionQueue to update
     119              : //! [`UploadQueueInitialized::visible_remote_consistent_lsn`] after it has
     120              : //! validated that our generation is not stale.  It is this visible value
     121              : //! that is advertized to safekeepers as a signal that that they can
     122              : //! delete the WAL up to that LSN.
     123              : //!
     124              : //! The [`RemoteTimelineClient::wait_completion`] method can be used to wait
     125              : //! for all pending operations to complete. It does not prevent more
     126              : //! operations from getting scheduled.
     127              : //!
     128              : //! # Crash Consistency
     129              : //!
     130              : //! We do not persist the upload queue state.
     131              : //! If we drop the client, or crash, all unfinished operations are lost.
     132              : //!
     133              : //! To recover, the following steps need to be taken:
     134              : //! - Retrieve the current remote [`IndexPart`]. This gives us a
     135              : //!   consistent remote state, assuming the user scheduled the operations in
     136              : //!   the correct order.
     137              : //! - Initiate upload queue with that [`IndexPart`].
     138              : //! - Reschedule all lost operations by comparing the local filesystem state
     139              : //!   and remote state as per [`IndexPart`]. This is done in
     140              : //!   [`Tenant::timeline_init_and_sync`].
     141              : //!
     142              : //! Note that if we crash during file deletion between the index update
     143              : //! that removes the file from the list of files, and deleting the remote file,
     144              : //! the file is leaked in the remote storage. Similarly, if a new file is created
     145              : //! and uploaded, but the pageserver dies permanently before updating the
     146              : //! remote index file, the new file is leaked in remote storage. We accept and
     147              : //! tolerate that for now.
     148              : //! Note further that we cannot easily fix this by scheduling deletes for every
     149              : //! file that is present only on the remote, because we cannot distinguish the
     150              : //! following two cases:
     151              : //! - (1) We had the file locally, deleted it locally, scheduled a remote delete,
     152              : //!   but crashed before it finished remotely.
     153              : //! - (2) We never had the file locally because we haven't on-demand downloaded
     154              : //!   it yet.
     155              : //!
     156              : //! # Downloads
     157              : //!
     158              : //! In addition to the upload queue, [`RemoteTimelineClient`] has functions for
     159              : //! downloading files from the remote storage. Downloads are performed immediately
     160              : //! against the `RemoteStorage`, independently of the upload queue.
     161              : //!
     162              : //! When we attach a tenant, we perform the following steps:
     163              : //! - create `Tenant` object in `TenantState::Attaching` state
     164              : //! - List timelines that are present in remote storage, and for each:
     165              : //!   - download their remote [`IndexPart`]s
     166              : //!   - create `Timeline` struct and a `RemoteTimelineClient`
     167              : //!   - initialize the client's upload queue with its `IndexPart`
     168              : //!   - schedule uploads for layers that are only present locally.
     169              : //! - After the above is done for each timeline, open the tenant for business by
     170              : //!   transitioning it from `TenantState::Attaching` to `TenantState::Active` state.
     171              : //!   This starts the timelines' WAL-receivers and the tenant's GC & Compaction loops.
     172              : //!
     173              : //! # Operating Without Remote Storage
     174              : //!
     175              : //! If no remote storage configuration is provided, the [`RemoteTimelineClient`] is
     176              : //! not created and the uploads are skipped.
     177              : //!
     178              : //! [`Tenant::timeline_init_and_sync`]: super::Tenant::timeline_init_and_sync
     179              : //! [`Timeline::load_layer_map`]: super::Timeline::load_layer_map
     180              : 
     181              : pub(crate) mod download;
     182              : pub mod index;
     183              : pub mod manifest;
     184              : pub(crate) mod upload;
     185              : 
     186              : use anyhow::Context;
     187              : use camino::Utf8Path;
     188              : use chrono::{NaiveDateTime, Utc};
     189              : 
     190              : pub(crate) use download::download_initdb_tar_zst;
     191              : use pageserver_api::models::TimelineArchivalState;
     192              : use pageserver_api::shard::{ShardIndex, TenantShardId};
     193              : use regex::Regex;
     194              : use scopeguard::ScopeGuard;
     195              : use tokio_util::sync::CancellationToken;
     196              : use utils::backoff::{
     197              :     self, exponential_backoff, DEFAULT_BASE_BACKOFF_SECONDS, DEFAULT_MAX_BACKOFF_SECONDS,
     198              : };
     199              : use utils::pausable_failpoint;
     200              : use utils::shard::ShardNumber;
     201              : 
     202              : use std::collections::{HashMap, HashSet, VecDeque};
     203              : use std::sync::atomic::{AtomicU32, Ordering};
     204              : use std::sync::{Arc, Mutex, OnceLock};
     205              : use std::time::Duration;
     206              : 
     207              : use remote_storage::{
     208              :     DownloadError, GenericRemoteStorage, ListingMode, RemotePath, TimeoutOrCancel,
     209              : };
     210              : use std::ops::DerefMut;
     211              : use tracing::{debug, error, info, instrument, warn};
     212              : use tracing::{info_span, Instrument};
     213              : use utils::lsn::Lsn;
     214              : 
     215              : use crate::context::RequestContext;
     216              : use crate::deletion_queue::{DeletionQueueClient, DeletionQueueError};
     217              : use crate::metrics::{
     218              :     MeasureRemoteOp, RemoteOpFileKind, RemoteOpKind, RemoteTimelineClientMetrics,
     219              :     RemoteTimelineClientMetricsCallTrackSize, REMOTE_ONDEMAND_DOWNLOADED_BYTES,
     220              :     REMOTE_ONDEMAND_DOWNLOADED_LAYERS,
     221              : };
     222              : use crate::task_mgr::shutdown_token;
     223              : use crate::tenant::debug_assert_current_span_has_tenant_and_timeline_id;
     224              : use crate::tenant::remote_timeline_client::download::download_retry;
     225              : use crate::tenant::storage_layer::AsLayerDesc;
     226              : use crate::tenant::upload_queue::{Delete, OpType, UploadQueueStoppedDeletable};
     227              : use crate::tenant::TIMELINES_SEGMENT_NAME;
     228              : use crate::{
     229              :     config::PageServerConf,
     230              :     task_mgr,
     231              :     task_mgr::TaskKind,
     232              :     task_mgr::BACKGROUND_RUNTIME,
     233              :     tenant::metadata::TimelineMetadata,
     234              :     tenant::upload_queue::{
     235              :         UploadOp, UploadQueue, UploadQueueInitialized, UploadQueueStopped, UploadTask,
     236              :     },
     237              :     TENANT_HEATMAP_BASENAME,
     238              : };
     239              : 
     240              : use utils::id::{TenantId, TimelineId};
     241              : 
     242              : use self::index::IndexPart;
     243              : 
     244              : use super::config::AttachedLocationConfig;
     245              : use super::metadata::MetadataUpdate;
     246              : use super::storage_layer::{Layer, LayerName, ResidentLayer};
     247              : use super::timeline::import_pgdata;
     248              : use super::upload_queue::{NotInitialized, SetDeletedFlagProgress};
     249              : use super::{DeleteTimelineError, Generation};
     250              : 
     251              : pub(crate) use download::{
     252              :     download_index_part, download_tenant_manifest, is_temp_download_file,
     253              :     list_remote_tenant_shards, list_remote_timelines,
     254              : };
     255              : pub(crate) use index::LayerFileMetadata;
     256              : pub(crate) use upload::upload_initdb_dir;
     257              : 
     258              : // Occasional network issues and such can cause remote operations to fail, and
     259              : // that's expected. If a download fails, we log it at info-level, and retry.
     260              : // But after FAILED_DOWNLOAD_WARN_THRESHOLD retries, we start to log it at WARN
     261              : // level instead, as repeated failures can mean a more serious problem. If it
     262              : // fails more than FAILED_DOWNLOAD_RETRIES times, we give up
     263              : pub(crate) const FAILED_DOWNLOAD_WARN_THRESHOLD: u32 = 3;
     264              : pub(crate) const FAILED_REMOTE_OP_RETRIES: u32 = 10;
     265              : 
     266              : // Similarly log failed uploads and deletions at WARN level, after this many
     267              : // retries. Uploads and deletions are retried forever, though.
     268              : pub(crate) const FAILED_UPLOAD_WARN_THRESHOLD: u32 = 3;
     269              : 
     270              : pub(crate) const INITDB_PATH: &str = "initdb.tar.zst";
     271              : 
     272              : pub(crate) const INITDB_PRESERVED_PATH: &str = "initdb-preserved.tar.zst";
     273              : 
     274              : /// Default buffer size when interfacing with [`tokio::fs::File`].
     275              : pub(crate) const BUFFER_SIZE: usize = 32 * 1024;
     276              : 
     277              : /// Doing non-essential flushes of deletion queue is subject to this timeout, after
     278              : /// which we warn and skip.
     279              : const DELETION_QUEUE_FLUSH_TIMEOUT: Duration = Duration::from_secs(10);
     280              : 
     281              : pub enum MaybeDeletedIndexPart {
     282              :     IndexPart(IndexPart),
     283              :     Deleted(IndexPart),
     284              : }
     285              : 
     286              : #[derive(Debug, thiserror::Error)]
     287              : pub enum PersistIndexPartWithDeletedFlagError {
     288              :     #[error("another task is already setting the deleted_flag, started at {0:?}")]
     289              :     AlreadyInProgress(NaiveDateTime),
     290              :     #[error("the deleted_flag was already set, value is {0:?}")]
     291              :     AlreadyDeleted(NaiveDateTime),
     292              :     #[error(transparent)]
     293              :     Other(#[from] anyhow::Error),
     294              : }
     295              : 
     296              : #[derive(Debug, thiserror::Error)]
     297              : pub enum WaitCompletionError {
     298              :     #[error(transparent)]
     299              :     NotInitialized(NotInitialized),
     300              :     #[error("wait_completion aborted because upload queue was stopped")]
     301              :     UploadQueueShutDownOrStopped,
     302              : }
     303              : 
     304              : #[derive(Debug, thiserror::Error)]
     305              : #[error("Upload queue either in unexpected state or hasn't downloaded manifest yet")]
     306              : pub struct UploadQueueNotReadyError;
     307              : /// Behavioral modes that enable seamless live migration.
     308              : ///
     309              : /// See docs/rfcs/028-pageserver-migration.md to understand how these fit in.
     310              : struct RemoteTimelineClientConfig {
     311              :     /// If this is false, then update to remote_consistent_lsn are dropped rather
     312              :     /// than being submitted to DeletionQueue for validation.  This behavior is
     313              :     /// used when a tenant attachment is known to have a stale generation number,
     314              :     /// such that validation attempts will always fail.  This is not necessary
     315              :     /// for correctness, but avoids spamming error statistics with failed validations
     316              :     /// when doing migrations of tenants.
     317              :     process_remote_consistent_lsn_updates: bool,
     318              : 
     319              :     /// If this is true, then object deletions are held in a buffer in RemoteTimelineClient
     320              :     /// rather than being submitted to the DeletionQueue.  This behavior is used when a tenant
     321              :     /// is known to be multi-attached, in order to avoid disrupting other attached tenants
     322              :     /// whose generations' metadata refers to the deleted objects.
     323              :     block_deletions: bool,
     324              : }
     325              : 
     326              : /// RemoteTimelineClientConfig's state is entirely driven by LocationConf, but we do
     327              : /// not carry the entire LocationConf structure: it's much more than we need.  The From
     328              : /// impl extracts the subset of the LocationConf that is interesting to RemoteTimelineClient.
     329              : impl From<&AttachedLocationConfig> for RemoteTimelineClientConfig {
     330          432 :     fn from(lc: &AttachedLocationConfig) -> Self {
     331          432 :         Self {
     332          432 :             block_deletions: !lc.may_delete_layers_hint(),
     333          432 :             process_remote_consistent_lsn_updates: lc.may_upload_layers_hint(),
     334          432 :         }
     335          432 :     }
     336              : }
     337              : 
     338              : /// A client for accessing a timeline's data in remote storage.
     339              : ///
     340              : /// This takes care of managing the number of connections, and balancing them
     341              : /// across tenants. This also handles retries of failed uploads.
     342              : ///
     343              : /// Upload and delete requests are ordered so that before a deletion is
     344              : /// performed, we wait for all preceding uploads to finish. This ensures sure
     345              : /// that if you perform a compaction operation that reshuffles data in layer
     346              : /// files, we don't have a transient state where the old files have already been
     347              : /// deleted, but new files have not yet been uploaded.
     348              : ///
     349              : /// Similarly, this enforces an order between index-file uploads, and layer
     350              : /// uploads.  Before an index-file upload is performed, all preceding layer
     351              : /// uploads must be finished.
     352              : ///
     353              : /// This also maintains a list of remote files, and automatically includes that
     354              : /// in the index part file, whenever timeline metadata is uploaded.
     355              : ///
     356              : /// Downloads are not queued, they are performed immediately.
     357              : pub(crate) struct RemoteTimelineClient {
     358              :     conf: &'static PageServerConf,
     359              : 
     360              :     runtime: tokio::runtime::Handle,
     361              : 
     362              :     tenant_shard_id: TenantShardId,
     363              :     timeline_id: TimelineId,
     364              :     generation: Generation,
     365              : 
     366              :     upload_queue: Mutex<UploadQueue>,
     367              : 
     368              :     pub(crate) metrics: Arc<RemoteTimelineClientMetrics>,
     369              : 
     370              :     storage_impl: GenericRemoteStorage,
     371              : 
     372              :     deletion_queue_client: DeletionQueueClient,
     373              : 
     374              :     /// Subset of tenant configuration used to control upload behaviors during migrations
     375              :     config: std::sync::RwLock<RemoteTimelineClientConfig>,
     376              : 
     377              :     cancel: CancellationToken,
     378              : }
     379              : 
     380              : impl RemoteTimelineClient {
     381              :     ///
     382              :     /// Create a remote storage client for given timeline
     383              :     ///
     384              :     /// Note: the caller must initialize the upload queue before any uploads can be scheduled,
     385              :     /// by calling init_upload_queue.
     386              :     ///
     387          422 :     pub(crate) fn new(
     388          422 :         remote_storage: GenericRemoteStorage,
     389          422 :         deletion_queue_client: DeletionQueueClient,
     390          422 :         conf: &'static PageServerConf,
     391          422 :         tenant_shard_id: TenantShardId,
     392          422 :         timeline_id: TimelineId,
     393          422 :         generation: Generation,
     394          422 :         location_conf: &AttachedLocationConfig,
     395          422 :     ) -> RemoteTimelineClient {
     396          422 :         RemoteTimelineClient {
     397          422 :             conf,
     398          422 :             runtime: if cfg!(test) {
     399              :                 // remote_timeline_client.rs tests rely on current-thread runtime
     400          422 :                 tokio::runtime::Handle::current()
     401              :             } else {
     402            0 :                 BACKGROUND_RUNTIME.handle().clone()
     403              :             },
     404          422 :             tenant_shard_id,
     405          422 :             timeline_id,
     406          422 :             generation,
     407          422 :             storage_impl: remote_storage,
     408          422 :             deletion_queue_client,
     409          422 :             upload_queue: Mutex::new(UploadQueue::Uninitialized),
     410          422 :             metrics: Arc::new(RemoteTimelineClientMetrics::new(
     411          422 :                 &tenant_shard_id,
     412          422 :                 &timeline_id,
     413          422 :             )),
     414          422 :             config: std::sync::RwLock::new(RemoteTimelineClientConfig::from(location_conf)),
     415          422 :             cancel: CancellationToken::new(),
     416          422 :         }
     417          422 :     }
     418              : 
     419              :     /// Initialize the upload queue for a remote storage that already received
     420              :     /// an index file upload, i.e., it's not empty.
     421              :     /// The given `index_part` must be the one on the remote.
     422            6 :     pub fn init_upload_queue(&self, index_part: &IndexPart) -> anyhow::Result<()> {
     423            6 :         let mut upload_queue = self.upload_queue.lock().unwrap();
     424            6 :         upload_queue.initialize_with_current_remote_index_part(index_part)?;
     425            6 :         self.update_remote_physical_size_gauge(Some(index_part));
     426            6 :         info!(
     427            0 :             "initialized upload queue from remote index with {} layer files",
     428            0 :             index_part.layer_metadata.len()
     429              :         );
     430            6 :         Ok(())
     431            6 :     }
     432              : 
     433              :     /// Initialize the upload queue for the case where the remote storage is empty,
     434              :     /// i.e., it doesn't have an `IndexPart`.
     435          416 :     pub fn init_upload_queue_for_empty_remote(
     436          416 :         &self,
     437          416 :         local_metadata: &TimelineMetadata,
     438          416 :     ) -> anyhow::Result<()> {
     439          416 :         let mut upload_queue = self.upload_queue.lock().unwrap();
     440          416 :         upload_queue.initialize_empty_remote(local_metadata)?;
     441          416 :         self.update_remote_physical_size_gauge(None);
     442          416 :         info!("initialized upload queue as empty");
     443          416 :         Ok(())
     444          416 :     }
     445              : 
     446              :     /// Initialize the queue in stopped state. Used in startup path
     447              :     /// to continue deletion operation interrupted by pageserver crash or restart.
     448            0 :     pub fn init_upload_queue_stopped_to_continue_deletion(
     449            0 :         &self,
     450            0 :         index_part: &IndexPart,
     451            0 :     ) -> anyhow::Result<()> {
     452              :         // FIXME: consider newtype for DeletedIndexPart.
     453            0 :         let deleted_at = index_part.deleted_at.ok_or(anyhow::anyhow!(
     454            0 :             "bug: it is responsibility of the caller to provide index part from MaybeDeletedIndexPart::Deleted"
     455            0 :         ))?;
     456              : 
     457            0 :         let mut upload_queue = self.upload_queue.lock().unwrap();
     458            0 :         upload_queue.initialize_with_current_remote_index_part(index_part)?;
     459            0 :         self.update_remote_physical_size_gauge(Some(index_part));
     460            0 :         self.stop_impl(&mut upload_queue);
     461            0 : 
     462            0 :         upload_queue
     463            0 :             .stopped_mut()
     464            0 :             .expect("stopped above")
     465            0 :             .deleted_at = SetDeletedFlagProgress::Successful(deleted_at);
     466            0 : 
     467            0 :         Ok(())
     468            0 :     }
     469              : 
     470              :     /// Notify this client of a change to its parent tenant's config, as this may cause us to
     471              :     /// take action (unblocking deletions when transitioning from AttachedMulti to AttachedSingle)
     472            0 :     pub(super) fn update_config(&self, location_conf: &AttachedLocationConfig) {
     473            0 :         let new_conf = RemoteTimelineClientConfig::from(location_conf);
     474            0 :         let unblocked = !new_conf.block_deletions;
     475            0 : 
     476            0 :         // Update config before draining deletions, so that we don't race with more being
     477            0 :         // inserted.  This can result in deletions happening our of order, but that does not
     478            0 :         // violate any invariants: deletions only need to be ordered relative to upload of the index
     479            0 :         // that dereferences the deleted objects, and we are not changing that order.
     480            0 :         *self.config.write().unwrap() = new_conf;
     481            0 : 
     482            0 :         if unblocked {
     483              :             // If we may now delete layers, drain any that were blocked in our old
     484              :             // configuration state
     485            0 :             let mut queue_locked = self.upload_queue.lock().unwrap();
     486              : 
     487            0 :             if let Ok(queue) = queue_locked.initialized_mut() {
     488            0 :                 let blocked_deletions = std::mem::take(&mut queue.blocked_deletions);
     489            0 :                 for d in blocked_deletions {
     490            0 :                     if let Err(e) = self.deletion_queue_client.push_layers_sync(
     491            0 :                         self.tenant_shard_id,
     492            0 :                         self.timeline_id,
     493            0 :                         self.generation,
     494            0 :                         d.layers,
     495            0 :                     ) {
     496              :                         // This could happen if the pageserver is shut down while a tenant
     497              :                         // is transitioning from a deletion-blocked state: we will leak some
     498              :                         // S3 objects in this case.
     499            0 :                         warn!("Failed to drain blocked deletions: {}", e);
     500            0 :                         break;
     501            0 :                     }
     502              :                 }
     503            0 :             }
     504            0 :         }
     505            0 :     }
     506              : 
     507              :     /// Returns `None` if nothing is yet uplodaded, `Some(disk_consistent_lsn)` otherwise.
     508            0 :     pub fn remote_consistent_lsn_projected(&self) -> Option<Lsn> {
     509            0 :         match &mut *self.upload_queue.lock().unwrap() {
     510            0 :             UploadQueue::Uninitialized => None,
     511            0 :             UploadQueue::Initialized(q) => q.get_last_remote_consistent_lsn_projected(),
     512            0 :             UploadQueue::Stopped(UploadQueueStopped::Uninitialized) => None,
     513            0 :             UploadQueue::Stopped(UploadQueueStopped::Deletable(q)) => q
     514            0 :                 .upload_queue_for_deletion
     515            0 :                 .get_last_remote_consistent_lsn_projected(),
     516              :         }
     517            0 :     }
     518              : 
     519            0 :     pub fn remote_consistent_lsn_visible(&self) -> Option<Lsn> {
     520            0 :         match &mut *self.upload_queue.lock().unwrap() {
     521            0 :             UploadQueue::Uninitialized => None,
     522            0 :             UploadQueue::Initialized(q) => Some(q.get_last_remote_consistent_lsn_visible()),
     523            0 :             UploadQueue::Stopped(UploadQueueStopped::Uninitialized) => None,
     524            0 :             UploadQueue::Stopped(UploadQueueStopped::Deletable(q)) => Some(
     525            0 :                 q.upload_queue_for_deletion
     526            0 :                     .get_last_remote_consistent_lsn_visible(),
     527            0 :             ),
     528              :         }
     529            0 :     }
     530              : 
     531              :     /// Returns true if this timeline was previously detached at this Lsn and the remote timeline
     532              :     /// client is currently initialized.
     533            0 :     pub(crate) fn is_previous_ancestor_lsn(&self, lsn: Lsn) -> bool {
     534            0 :         self.upload_queue
     535            0 :             .lock()
     536            0 :             .unwrap()
     537            0 :             .initialized_mut()
     538            0 :             .map(|uq| uq.clean.0.lineage.is_previous_ancestor_lsn(lsn))
     539            0 :             .unwrap_or(false)
     540            0 :     }
     541              : 
     542              :     /// Returns whether the timeline is archived.
     543              :     /// Return None if the remote index_part hasn't been downloaded yet.
     544            2 :     pub(crate) fn is_archived(&self) -> Option<bool> {
     545            2 :         self.upload_queue
     546            2 :             .lock()
     547            2 :             .unwrap()
     548            2 :             .initialized_mut()
     549            2 :             .map(|q| q.clean.0.archived_at.is_some())
     550            2 :             .ok()
     551            2 :     }
     552              : 
     553              :     /// Returns `Ok(Some(timestamp))` if the timeline has been archived, `Ok(None)` if the timeline hasn't been archived.
     554              :     ///
     555              :     /// Return Err(_) if the remote index_part hasn't been downloaded yet, or the timeline hasn't been stopped yet.
     556            2 :     pub(crate) fn archived_at_stopped_queue(
     557            2 :         &self,
     558            2 :     ) -> Result<Option<NaiveDateTime>, UploadQueueNotReadyError> {
     559            2 :         self.upload_queue
     560            2 :             .lock()
     561            2 :             .unwrap()
     562            2 :             .stopped_mut()
     563            2 :             .map(|q| q.upload_queue_for_deletion.clean.0.archived_at)
     564            2 :             .map_err(|_| UploadQueueNotReadyError)
     565            2 :     }
     566              : 
     567         1878 :     fn update_remote_physical_size_gauge(&self, current_remote_index_part: Option<&IndexPart>) {
     568         1878 :         let size: u64 = if let Some(current_remote_index_part) = current_remote_index_part {
     569         1462 :             current_remote_index_part
     570         1462 :                 .layer_metadata
     571         1462 :                 .values()
     572        17597 :                 .map(|ilmd| ilmd.file_size)
     573         1462 :                 .sum()
     574              :         } else {
     575          416 :             0
     576              :         };
     577         1878 :         self.metrics.remote_physical_size_gauge.set(size);
     578         1878 :     }
     579              : 
     580            2 :     pub fn get_remote_physical_size(&self) -> u64 {
     581            2 :         self.metrics.remote_physical_size_gauge.get()
     582            2 :     }
     583              : 
     584              :     //
     585              :     // Download operations.
     586              :     //
     587              :     // These don't use the per-timeline queue. They do use the global semaphore in
     588              :     // S3Bucket, to limit the total number of concurrent operations, though.
     589              :     //
     590              : 
     591              :     /// Download index file
     592           20 :     pub async fn download_index_file(
     593           20 :         &self,
     594           20 :         cancel: &CancellationToken,
     595           20 :     ) -> Result<MaybeDeletedIndexPart, DownloadError> {
     596           20 :         let _unfinished_gauge_guard = self.metrics.call_begin(
     597           20 :             &RemoteOpFileKind::Index,
     598           20 :             &RemoteOpKind::Download,
     599           20 :             crate::metrics::RemoteTimelineClientMetricsCallTrackSize::DontTrackSize {
     600           20 :                 reason: "no need for a downloads gauge",
     601           20 :             },
     602           20 :         );
     603              : 
     604           20 :         let (index_part, index_generation, index_last_modified) = download::download_index_part(
     605           20 :             &self.storage_impl,
     606           20 :             &self.tenant_shard_id,
     607           20 :             &self.timeline_id,
     608           20 :             self.generation,
     609           20 :             cancel,
     610           20 :         )
     611           20 :         .measure_remote_op(
     612           20 :             RemoteOpFileKind::Index,
     613           20 :             RemoteOpKind::Download,
     614           20 :             Arc::clone(&self.metrics),
     615           20 :         )
     616           20 :         .await?;
     617              : 
     618              :         // Defense in depth: monotonicity of generation numbers is an important correctness guarantee, so when we see a very
     619              :         // old index, we do extra checks in case this is the result of backward time-travel of the generation number (e.g.
     620              :         // in case of a bug in the service that issues generation numbers). Indices are allowed to be old, but we expect that
     621              :         // when we load an old index we are loading the _latest_ index: if we are asked to load an old index and there is
     622              :         // also a newer index available, that is surprising.
     623              :         const INDEX_AGE_CHECKS_THRESHOLD: Duration = Duration::from_secs(14 * 24 * 3600);
     624           20 :         let index_age = index_last_modified.elapsed().unwrap_or_else(|e| {
     625            0 :             if e.duration() > Duration::from_secs(5) {
     626              :                 // We only warn if the S3 clock and our local clock are >5s out: because this is a low resolution
     627              :                 // timestamp, it is common to be out by at least 1 second.
     628            0 :                 tracing::warn!("Index has modification time in the future: {e}");
     629            0 :             }
     630            0 :             Duration::ZERO
     631           20 :         });
     632           20 :         if index_age > INDEX_AGE_CHECKS_THRESHOLD {
     633            0 :             tracing::info!(
     634              :                 ?index_generation,
     635            0 :                 age = index_age.as_secs_f64(),
     636            0 :                 "Loaded an old index, checking for other indices..."
     637              :             );
     638              : 
     639              :             // Find the highest-generation index
     640            0 :             let (_latest_index_part, latest_index_generation, latest_index_mtime) =
     641            0 :                 download::download_index_part(
     642            0 :                     &self.storage_impl,
     643            0 :                     &self.tenant_shard_id,
     644            0 :                     &self.timeline_id,
     645            0 :                     Generation::MAX,
     646            0 :                     cancel,
     647            0 :                 )
     648            0 :                 .await?;
     649              : 
     650            0 :             if latest_index_generation > index_generation {
     651              :                 // Unexpected!  Why are we loading such an old index if a more recent one exists?
     652              :                 // We will refuse to proceed, as there is no reasonable scenario where this should happen, but
     653              :                 // there _is_ a clear bug/corruption scenario where it would happen (controller sets the generation
     654              :                 // backwards).
     655            0 :                 tracing::error!(
     656              :                     ?index_generation,
     657              :                     ?latest_index_generation,
     658              :                     ?latest_index_mtime,
     659            0 :                     "Found a newer index while loading an old one"
     660              :                 );
     661            0 :                 return Err(DownloadError::Fatal(
     662            0 :                     "Index age exceeds threshold and a newer index exists".into(),
     663            0 :                 ));
     664            0 :             }
     665           20 :         }
     666              : 
     667           20 :         if index_part.deleted_at.is_some() {
     668            0 :             Ok(MaybeDeletedIndexPart::Deleted(index_part))
     669              :         } else {
     670           20 :             Ok(MaybeDeletedIndexPart::IndexPart(index_part))
     671              :         }
     672           20 :     }
     673              : 
     674              :     /// Download a (layer) file from `path`, into local filesystem.
     675              :     ///
     676              :     /// 'layer_metadata' is the metadata from the remote index file.
     677              :     ///
     678              :     /// On success, returns the size of the downloaded file.
     679            6 :     pub async fn download_layer_file(
     680            6 :         &self,
     681            6 :         layer_file_name: &LayerName,
     682            6 :         layer_metadata: &LayerFileMetadata,
     683            6 :         local_path: &Utf8Path,
     684            6 :         gate: &utils::sync::gate::Gate,
     685            6 :         cancel: &CancellationToken,
     686            6 :         ctx: &RequestContext,
     687            6 :     ) -> Result<u64, DownloadError> {
     688            6 :         let downloaded_size = {
     689            6 :             let _unfinished_gauge_guard = self.metrics.call_begin(
     690            6 :                 &RemoteOpFileKind::Layer,
     691            6 :                 &RemoteOpKind::Download,
     692            6 :                 crate::metrics::RemoteTimelineClientMetricsCallTrackSize::DontTrackSize {
     693            6 :                     reason: "no need for a downloads gauge",
     694            6 :                 },
     695            6 :             );
     696            6 :             download::download_layer_file(
     697            6 :                 self.conf,
     698            6 :                 &self.storage_impl,
     699            6 :                 self.tenant_shard_id,
     700            6 :                 self.timeline_id,
     701            6 :                 layer_file_name,
     702            6 :                 layer_metadata,
     703            6 :                 local_path,
     704            6 :                 gate,
     705            6 :                 cancel,
     706            6 :                 ctx,
     707            6 :             )
     708            6 :             .measure_remote_op(
     709            6 :                 RemoteOpFileKind::Layer,
     710            6 :                 RemoteOpKind::Download,
     711            6 :                 Arc::clone(&self.metrics),
     712            6 :             )
     713            6 :             .await?
     714              :         };
     715              : 
     716            6 :         REMOTE_ONDEMAND_DOWNLOADED_LAYERS.inc();
     717            6 :         REMOTE_ONDEMAND_DOWNLOADED_BYTES.inc_by(downloaded_size);
     718            6 : 
     719            6 :         Ok(downloaded_size)
     720            6 :     }
     721              : 
     722              :     //
     723              :     // Upload operations.
     724              :     //
     725              : 
     726              :     /// Launch an index-file upload operation in the background, with
     727              :     /// fully updated metadata.
     728              :     ///
     729              :     /// This should only be used to upload initial metadata to remote storage.
     730              :     ///
     731              :     /// The upload will be added to the queue immediately, but it
     732              :     /// won't be performed until all previously scheduled layer file
     733              :     /// upload operations have completed successfully.  This is to
     734              :     /// ensure that when the index file claims that layers X, Y and Z
     735              :     /// exist in remote storage, they really do. To wait for the upload
     736              :     /// to complete, use `wait_completion`.
     737              :     ///
     738              :     /// If there were any changes to the list of files, i.e. if any
     739              :     /// layer file uploads were scheduled, since the last index file
     740              :     /// upload, those will be included too.
     741          230 :     pub fn schedule_index_upload_for_full_metadata_update(
     742          230 :         self: &Arc<Self>,
     743          230 :         metadata: &TimelineMetadata,
     744          230 :     ) -> anyhow::Result<()> {
     745          230 :         let mut guard = self.upload_queue.lock().unwrap();
     746          230 :         let upload_queue = guard.initialized_mut()?;
     747              : 
     748              :         // As documented in the struct definition, it's ok for latest_metadata to be
     749              :         // ahead of what's _actually_ on the remote during index upload.
     750          230 :         upload_queue.dirty.metadata = metadata.clone();
     751          230 : 
     752          230 :         self.schedule_index_upload(upload_queue);
     753          230 : 
     754          230 :         Ok(())
     755          230 :     }
     756              : 
     757              :     /// Launch an index-file upload operation in the background, with only parts of the metadata
     758              :     /// updated.
     759              :     ///
     760              :     /// This is the regular way of updating metadata on layer flushes or Gc.
     761              :     ///
     762              :     /// Using this lighter update mechanism allows for reparenting and detaching without changes to
     763              :     /// `index_part.json`, while being more clear on what values update regularly.
     764         1198 :     pub(crate) fn schedule_index_upload_for_metadata_update(
     765         1198 :         self: &Arc<Self>,
     766         1198 :         update: &MetadataUpdate,
     767         1198 :     ) -> anyhow::Result<()> {
     768         1198 :         let mut guard = self.upload_queue.lock().unwrap();
     769         1198 :         let upload_queue = guard.initialized_mut()?;
     770              : 
     771         1198 :         upload_queue.dirty.metadata.apply(update);
     772         1198 : 
     773         1198 :         self.schedule_index_upload(upload_queue);
     774         1198 : 
     775         1198 :         Ok(())
     776         1198 :     }
     777              : 
     778              :     /// Launch an index-file upload operation in the background, with only the `archived_at` field updated.
     779              :     ///
     780              :     /// Returns whether it is required to wait for the queue to be empty to ensure that the change is uploaded,
     781              :     /// so either if the change is already sitting in the queue, but not commited yet, or the change has not
     782              :     /// been in the queue yet.
     783            2 :     pub(crate) fn schedule_index_upload_for_timeline_archival_state(
     784            2 :         self: &Arc<Self>,
     785            2 :         state: TimelineArchivalState,
     786            2 :     ) -> anyhow::Result<bool> {
     787            2 :         let mut guard = self.upload_queue.lock().unwrap();
     788            2 :         let upload_queue = guard.initialized_mut()?;
     789              : 
     790              :         /// Returns Some(_) if a change is needed, and Some(true) if it's a
     791              :         /// change needed to set archived_at.
     792            4 :         fn need_change(
     793            4 :             archived_at: &Option<NaiveDateTime>,
     794            4 :             state: TimelineArchivalState,
     795            4 :         ) -> Option<bool> {
     796            4 :             match (archived_at, state) {
     797              :                 (Some(_), TimelineArchivalState::Archived)
     798              :                 | (None, TimelineArchivalState::Unarchived) => {
     799              :                     // Nothing to do
     800            0 :                     tracing::info!("intended state matches present state");
     801            0 :                     None
     802              :                 }
     803            4 :                 (None, TimelineArchivalState::Archived) => Some(true),
     804            0 :                 (Some(_), TimelineArchivalState::Unarchived) => Some(false),
     805              :             }
     806            4 :         }
     807            2 :         let need_upload_scheduled = need_change(&upload_queue.dirty.archived_at, state);
     808              : 
     809            2 :         if let Some(archived_at_set) = need_upload_scheduled {
     810            2 :             let intended_archived_at = archived_at_set.then(|| Utc::now().naive_utc());
     811            2 :             upload_queue.dirty.archived_at = intended_archived_at;
     812            2 :             self.schedule_index_upload(upload_queue);
     813            2 :         }
     814              : 
     815            2 :         let need_wait = need_change(&upload_queue.clean.0.archived_at, state).is_some();
     816            2 :         Ok(need_wait)
     817            2 :     }
     818              : 
     819              :     /// Launch an index-file upload operation in the background, setting `import_pgdata` field.
     820            0 :     pub(crate) fn schedule_index_upload_for_import_pgdata_state_update(
     821            0 :         self: &Arc<Self>,
     822            0 :         state: Option<import_pgdata::index_part_format::Root>,
     823            0 :     ) -> anyhow::Result<()> {
     824            0 :         let mut guard = self.upload_queue.lock().unwrap();
     825            0 :         let upload_queue = guard.initialized_mut()?;
     826            0 :         upload_queue.dirty.import_pgdata = state;
     827            0 :         self.schedule_index_upload(upload_queue);
     828            0 :         Ok(())
     829            0 :     }
     830              : 
     831              :     ///
     832              :     /// Launch an index-file upload operation in the background, if necessary.
     833              :     ///
     834              :     /// Use this function to schedule the update of the index file after
     835              :     /// scheduling file uploads or deletions. If no file uploads or deletions
     836              :     /// have been scheduled since the last index file upload, this does
     837              :     /// nothing.
     838              :     ///
     839              :     /// Like schedule_index_upload_for_metadata_update(), this merely adds
     840              :     /// the upload to the upload queue and returns quickly.
     841          370 :     pub fn schedule_index_upload_for_file_changes(self: &Arc<Self>) -> Result<(), NotInitialized> {
     842          370 :         let mut guard = self.upload_queue.lock().unwrap();
     843          370 :         let upload_queue = guard.initialized_mut()?;
     844              : 
     845          370 :         if upload_queue.latest_files_changes_since_metadata_upload_scheduled > 0 {
     846           14 :             self.schedule_index_upload(upload_queue);
     847          356 :         }
     848              : 
     849          370 :         Ok(())
     850          370 :     }
     851              : 
     852              :     /// Launch an index-file upload operation in the background (internal function)
     853         1514 :     fn schedule_index_upload(self: &Arc<Self>, upload_queue: &mut UploadQueueInitialized) {
     854         1514 :         let disk_consistent_lsn = upload_queue.dirty.metadata.disk_consistent_lsn();
     855         1514 :         // fix up the duplicated field
     856         1514 :         upload_queue.dirty.disk_consistent_lsn = disk_consistent_lsn;
     857         1514 : 
     858         1514 :         // make sure it serializes before doing it in perform_upload_task so that it doesn't
     859         1514 :         // look like a retryable error
     860         1514 :         let void = std::io::sink();
     861         1514 :         serde_json::to_writer(void, &upload_queue.dirty).expect("serialize index_part.json");
     862         1514 : 
     863         1514 :         let index_part = &upload_queue.dirty;
     864         1514 : 
     865         1514 :         info!(
     866            0 :             "scheduling metadata upload up to consistent LSN {disk_consistent_lsn} with {} files ({} changed)",
     867            0 :             index_part.layer_metadata.len(),
     868              :             upload_queue.latest_files_changes_since_metadata_upload_scheduled,
     869              :         );
     870              : 
     871         1514 :         let op = UploadOp::UploadMetadata {
     872         1514 :             uploaded: Box::new(index_part.clone()),
     873         1514 :         };
     874         1514 :         self.metric_begin(&op);
     875         1514 :         upload_queue.queued_operations.push_back(op);
     876         1514 :         upload_queue.latest_files_changes_since_metadata_upload_scheduled = 0;
     877         1514 : 
     878         1514 :         // Launch the task immediately, if possible
     879         1514 :         self.launch_queued_tasks(upload_queue);
     880         1514 :     }
     881              : 
     882              :     /// Reparent this timeline to a new parent.
     883              :     ///
     884              :     /// A retryable step of timeline ancestor detach.
     885            0 :     pub(crate) async fn schedule_reparenting_and_wait(
     886            0 :         self: &Arc<Self>,
     887            0 :         new_parent: &TimelineId,
     888            0 :     ) -> anyhow::Result<()> {
     889            0 :         let receiver = {
     890            0 :             let mut guard = self.upload_queue.lock().unwrap();
     891            0 :             let upload_queue = guard.initialized_mut()?;
     892              : 
     893            0 :             let Some(prev) = upload_queue.dirty.metadata.ancestor_timeline() else {
     894            0 :                 return Err(anyhow::anyhow!(
     895            0 :                     "cannot reparent without a current ancestor"
     896            0 :                 ));
     897              :             };
     898              : 
     899            0 :             let uploaded = &upload_queue.clean.0.metadata;
     900            0 : 
     901            0 :             if uploaded.ancestor_timeline().is_none() && !uploaded.ancestor_lsn().is_valid() {
     902              :                 // nothing to do
     903            0 :                 None
     904              :             } else {
     905            0 :                 upload_queue.dirty.metadata.reparent(new_parent);
     906            0 :                 upload_queue.dirty.lineage.record_previous_ancestor(&prev);
     907            0 : 
     908            0 :                 self.schedule_index_upload(upload_queue);
     909            0 : 
     910            0 :                 Some(self.schedule_barrier0(upload_queue))
     911              :             }
     912              :         };
     913              : 
     914            0 :         if let Some(receiver) = receiver {
     915            0 :             Self::wait_completion0(receiver).await?;
     916            0 :         }
     917            0 :         Ok(())
     918            0 :     }
     919              : 
     920              :     /// Schedules uploading a new version of `index_part.json` with the given layers added,
     921              :     /// detaching from ancestor and waits for it to complete.
     922              :     ///
     923              :     /// This is used with `Timeline::detach_ancestor` functionality.
     924            0 :     pub(crate) async fn schedule_adding_existing_layers_to_index_detach_and_wait(
     925            0 :         self: &Arc<Self>,
     926            0 :         layers: &[Layer],
     927            0 :         adopted: (TimelineId, Lsn),
     928            0 :     ) -> anyhow::Result<()> {
     929            0 :         let barrier = {
     930            0 :             let mut guard = self.upload_queue.lock().unwrap();
     931            0 :             let upload_queue = guard.initialized_mut()?;
     932              : 
     933            0 :             if upload_queue.clean.0.lineage.detached_previous_ancestor() == Some(adopted) {
     934            0 :                 None
     935              :             } else {
     936            0 :                 upload_queue.dirty.metadata.detach_from_ancestor(&adopted);
     937            0 :                 upload_queue.dirty.lineage.record_detaching(&adopted);
     938              : 
     939            0 :                 for layer in layers {
     940            0 :                     let prev = upload_queue
     941            0 :                         .dirty
     942            0 :                         .layer_metadata
     943            0 :                         .insert(layer.layer_desc().layer_name(), layer.metadata());
     944            0 :                     assert!(prev.is_none(), "copied layer existed already {layer}");
     945              :                 }
     946              : 
     947            0 :                 self.schedule_index_upload(upload_queue);
     948            0 : 
     949            0 :                 Some(self.schedule_barrier0(upload_queue))
     950              :             }
     951              :         };
     952              : 
     953            0 :         if let Some(barrier) = barrier {
     954            0 :             Self::wait_completion0(barrier).await?;
     955            0 :         }
     956            0 :         Ok(())
     957            0 :     }
     958              : 
     959              :     /// Adds a gc blocking reason for this timeline if one does not exist already.
     960              :     ///
     961              :     /// A retryable step of timeline detach ancestor.
     962              :     ///
     963              :     /// Returns a future which waits until the completion of the upload.
     964            0 :     pub(crate) fn schedule_insert_gc_block_reason(
     965            0 :         self: &Arc<Self>,
     966            0 :         reason: index::GcBlockingReason,
     967            0 :     ) -> Result<impl std::future::Future<Output = Result<(), WaitCompletionError>>, NotInitialized>
     968            0 :     {
     969            0 :         let maybe_barrier = {
     970            0 :             let mut guard = self.upload_queue.lock().unwrap();
     971            0 :             let upload_queue = guard.initialized_mut()?;
     972              : 
     973            0 :             if let index::GcBlockingReason::DetachAncestor = reason {
     974            0 :                 if upload_queue.dirty.metadata.ancestor_timeline().is_none() {
     975            0 :                     drop(guard);
     976            0 :                     panic!("cannot start detach ancestor if there is nothing to detach from");
     977            0 :                 }
     978            0 :             }
     979              : 
     980            0 :             let wanted = |x: Option<&index::GcBlocking>| x.is_some_and(|x| x.blocked_by(reason));
     981              : 
     982            0 :             let current = upload_queue.dirty.gc_blocking.as_ref();
     983            0 :             let uploaded = upload_queue.clean.0.gc_blocking.as_ref();
     984            0 : 
     985            0 :             match (current, uploaded) {
     986            0 :                 (x, y) if wanted(x) && wanted(y) => None,
     987            0 :                 (x, y) if wanted(x) && !wanted(y) => Some(self.schedule_barrier0(upload_queue)),
     988              :                 // Usual case: !wanted(x) && !wanted(y)
     989              :                 //
     990              :                 // Unusual: !wanted(x) && wanted(y) which means we have two processes waiting to
     991              :                 // turn on and off some reason.
     992            0 :                 (x, y) => {
     993            0 :                     if !wanted(x) && wanted(y) {
     994              :                         // this could be avoided by having external in-memory synchronization, like
     995              :                         // timeline detach ancestor
     996            0 :                         warn!(?reason, op="insert", "unexpected: two racing processes to enable and disable a gc blocking reason");
     997            0 :                     }
     998              : 
     999              :                     // at this point, the metadata must always show that there is a parent
    1000            0 :                     upload_queue.dirty.gc_blocking = current
    1001            0 :                         .map(|x| x.with_reason(reason))
    1002            0 :                         .or_else(|| Some(index::GcBlocking::started_now_for(reason)));
    1003            0 :                     self.schedule_index_upload(upload_queue);
    1004            0 :                     Some(self.schedule_barrier0(upload_queue))
    1005              :                 }
    1006              :             }
    1007              :         };
    1008              : 
    1009            0 :         Ok(async move {
    1010            0 :             if let Some(barrier) = maybe_barrier {
    1011            0 :                 Self::wait_completion0(barrier).await?;
    1012            0 :             }
    1013            0 :             Ok(())
    1014            0 :         })
    1015            0 :     }
    1016              : 
    1017              :     /// Removes a gc blocking reason for this timeline if one exists.
    1018              :     ///
    1019              :     /// A retryable step of timeline detach ancestor.
    1020              :     ///
    1021              :     /// Returns a future which waits until the completion of the upload.
    1022            0 :     pub(crate) fn schedule_remove_gc_block_reason(
    1023            0 :         self: &Arc<Self>,
    1024            0 :         reason: index::GcBlockingReason,
    1025            0 :     ) -> Result<impl std::future::Future<Output = Result<(), WaitCompletionError>>, NotInitialized>
    1026            0 :     {
    1027            0 :         let maybe_barrier = {
    1028            0 :             let mut guard = self.upload_queue.lock().unwrap();
    1029            0 :             let upload_queue = guard.initialized_mut()?;
    1030              : 
    1031            0 :             if let index::GcBlockingReason::DetachAncestor = reason {
    1032            0 :                 if !upload_queue.clean.0.lineage.is_detached_from_ancestor() {
    1033            0 :                     drop(guard);
    1034            0 :                     panic!("cannot complete timeline_ancestor_detach while not detached");
    1035            0 :                 }
    1036            0 :             }
    1037              : 
    1038            0 :             let wanted = |x: Option<&index::GcBlocking>| {
    1039            0 :                 x.is_none() || x.is_some_and(|b| !b.blocked_by(reason))
    1040            0 :             };
    1041              : 
    1042            0 :             let current = upload_queue.dirty.gc_blocking.as_ref();
    1043            0 :             let uploaded = upload_queue.clean.0.gc_blocking.as_ref();
    1044            0 : 
    1045            0 :             match (current, uploaded) {
    1046            0 :                 (x, y) if wanted(x) && wanted(y) => None,
    1047            0 :                 (x, y) if wanted(x) && !wanted(y) => Some(self.schedule_barrier0(upload_queue)),
    1048            0 :                 (x, y) => {
    1049            0 :                     if !wanted(x) && wanted(y) {
    1050            0 :                         warn!(?reason, op="remove", "unexpected: two racing processes to enable and disable a gc blocking reason (remove)");
    1051            0 :                     }
    1052              : 
    1053            0 :                     upload_queue.dirty.gc_blocking =
    1054            0 :                         current.as_ref().and_then(|x| x.without_reason(reason));
    1055            0 :                     assert!(wanted(upload_queue.dirty.gc_blocking.as_ref()));
    1056            0 :                     self.schedule_index_upload(upload_queue);
    1057            0 :                     Some(self.schedule_barrier0(upload_queue))
    1058              :                 }
    1059              :             }
    1060              :         };
    1061              : 
    1062            0 :         Ok(async move {
    1063            0 :             if let Some(barrier) = maybe_barrier {
    1064            0 :                 Self::wait_completion0(barrier).await?;
    1065            0 :             }
    1066            0 :             Ok(())
    1067            0 :         })
    1068            0 :     }
    1069              : 
    1070              :     /// Launch an upload operation in the background; the file is added to be included in next
    1071              :     /// `index_part.json` upload.
    1072         1194 :     pub(crate) fn schedule_layer_file_upload(
    1073         1194 :         self: &Arc<Self>,
    1074         1194 :         layer: ResidentLayer,
    1075         1194 :     ) -> Result<(), NotInitialized> {
    1076         1194 :         let mut guard = self.upload_queue.lock().unwrap();
    1077         1194 :         let upload_queue = guard.initialized_mut()?;
    1078              : 
    1079         1194 :         self.schedule_layer_file_upload0(upload_queue, layer);
    1080         1194 :         self.launch_queued_tasks(upload_queue);
    1081         1194 :         Ok(())
    1082         1194 :     }
    1083              : 
    1084         1556 :     fn schedule_layer_file_upload0(
    1085         1556 :         self: &Arc<Self>,
    1086         1556 :         upload_queue: &mut UploadQueueInitialized,
    1087         1556 :         layer: ResidentLayer,
    1088         1556 :     ) {
    1089         1556 :         let metadata = layer.metadata();
    1090         1556 : 
    1091         1556 :         upload_queue
    1092         1556 :             .dirty
    1093         1556 :             .layer_metadata
    1094         1556 :             .insert(layer.layer_desc().layer_name(), metadata.clone());
    1095         1556 :         upload_queue.latest_files_changes_since_metadata_upload_scheduled += 1;
    1096         1556 : 
    1097         1556 :         info!(
    1098              :             gen=?metadata.generation,
    1099              :             shard=?metadata.shard,
    1100            0 :             "scheduled layer file upload {layer}",
    1101              :         );
    1102              : 
    1103         1556 :         let op = UploadOp::UploadLayer(layer, metadata, None);
    1104         1556 :         self.metric_begin(&op);
    1105         1556 :         upload_queue.queued_operations.push_back(op);
    1106         1556 :     }
    1107              : 
    1108              :     /// Launch a delete operation in the background.
    1109              :     ///
    1110              :     /// The operation does not modify local filesystem state.
    1111              :     ///
    1112              :     /// Note: This schedules an index file upload before the deletions.  The
    1113              :     /// deletion won't actually be performed, until all previously scheduled
    1114              :     /// upload operations, and the index file upload, have completed
    1115              :     /// successfully.
    1116            8 :     pub fn schedule_layer_file_deletion(
    1117            8 :         self: &Arc<Self>,
    1118            8 :         names: &[LayerName],
    1119            8 :     ) -> anyhow::Result<()> {
    1120            8 :         let mut guard = self.upload_queue.lock().unwrap();
    1121            8 :         let upload_queue = guard.initialized_mut()?;
    1122              : 
    1123            8 :         let with_metadata =
    1124            8 :             self.schedule_unlinking_of_layers_from_index_part0(upload_queue, names.iter().cloned());
    1125            8 : 
    1126            8 :         self.schedule_deletion_of_unlinked0(upload_queue, with_metadata);
    1127            8 : 
    1128            8 :         // Launch the tasks immediately, if possible
    1129            8 :         self.launch_queued_tasks(upload_queue);
    1130            8 :         Ok(())
    1131            8 :     }
    1132              : 
    1133              :     /// Unlinks the layer files from `index_part.json` but does not yet schedule deletion for the
    1134              :     /// layer files, leaving them dangling.
    1135              :     ///
    1136              :     /// The files will be leaked in remote storage unless [`Self::schedule_deletion_of_unlinked`]
    1137              :     /// is invoked on them.
    1138            4 :     pub(crate) fn schedule_gc_update(
    1139            4 :         self: &Arc<Self>,
    1140            4 :         gc_layers: &[Layer],
    1141            4 :     ) -> Result<(), NotInitialized> {
    1142            4 :         let mut guard = self.upload_queue.lock().unwrap();
    1143            4 :         let upload_queue = guard.initialized_mut()?;
    1144              : 
    1145              :         // just forget the return value; after uploading the next index_part.json, we can consider
    1146              :         // the layer files as "dangling". this is fine, at worst case we create work for the
    1147              :         // scrubber.
    1148              : 
    1149            4 :         let names = gc_layers.iter().map(|x| x.layer_desc().layer_name());
    1150            4 : 
    1151            4 :         self.schedule_unlinking_of_layers_from_index_part0(upload_queue, names);
    1152            4 : 
    1153            4 :         self.launch_queued_tasks(upload_queue);
    1154            4 : 
    1155            4 :         Ok(())
    1156            4 :     }
    1157              : 
    1158              :     /// Update the remote index file, removing the to-be-deleted files from the index,
    1159              :     /// allowing scheduling of actual deletions later.
    1160           88 :     fn schedule_unlinking_of_layers_from_index_part0<I>(
    1161           88 :         self: &Arc<Self>,
    1162           88 :         upload_queue: &mut UploadQueueInitialized,
    1163           88 :         names: I,
    1164           88 :     ) -> Vec<(LayerName, LayerFileMetadata)>
    1165           88 :     where
    1166           88 :         I: IntoIterator<Item = LayerName>,
    1167           88 :     {
    1168           88 :         // Decorate our list of names with each name's metadata, dropping
    1169           88 :         // names that are unexpectedly missing from our metadata.  This metadata
    1170           88 :         // is later used when physically deleting layers, to construct key paths.
    1171           88 :         let with_metadata: Vec<_> = names
    1172           88 :             .into_iter()
    1173          512 :             .filter_map(|name| {
    1174          512 :                 let meta = upload_queue.dirty.layer_metadata.remove(&name);
    1175              : 
    1176          512 :                 if let Some(meta) = meta {
    1177          446 :                     upload_queue.latest_files_changes_since_metadata_upload_scheduled += 1;
    1178          446 :                     Some((name, meta))
    1179              :                 } else {
    1180              :                     // This can only happen if we forgot to to schedule the file upload
    1181              :                     // before scheduling the delete. Log it because it is a rare/strange
    1182              :                     // situation, and in case something is misbehaving, we'd like to know which
    1183              :                     // layers experienced this.
    1184           66 :                     info!("Deleting layer {name} not found in latest_files list, never uploaded?");
    1185           66 :                     None
    1186              :                 }
    1187          512 :             })
    1188           88 :             .collect();
    1189              : 
    1190              :         #[cfg(feature = "testing")]
    1191          534 :         for (name, metadata) in &with_metadata {
    1192          446 :             let gen = metadata.generation;
    1193          446 :             if let Some(unexpected) = upload_queue.dangling_files.insert(name.to_owned(), gen) {
    1194            0 :                 if unexpected == gen {
    1195            0 :                     tracing::error!("{name} was unlinked twice with same generation");
    1196              :                 } else {
    1197            0 :                     tracing::error!("{name} was unlinked twice with different generations {gen:?} and {unexpected:?}");
    1198              :                 }
    1199          446 :             }
    1200              :         }
    1201              : 
    1202              :         // after unlinking files from the upload_queue.latest_files we must always schedule an
    1203              :         // index_part update, because that needs to be uploaded before we can actually delete the
    1204              :         // files.
    1205           88 :         if upload_queue.latest_files_changes_since_metadata_upload_scheduled > 0 {
    1206           70 :             self.schedule_index_upload(upload_queue);
    1207           70 :         }
    1208              : 
    1209           88 :         with_metadata
    1210           88 :     }
    1211              : 
    1212              :     /// Schedules deletion for layer files which have previously been unlinked from the
    1213              :     /// `index_part.json` with [`Self::schedule_gc_update`] or [`Self::schedule_compaction_update`].
    1214          511 :     pub(crate) fn schedule_deletion_of_unlinked(
    1215          511 :         self: &Arc<Self>,
    1216          511 :         layers: Vec<(LayerName, LayerFileMetadata)>,
    1217          511 :     ) -> anyhow::Result<()> {
    1218          511 :         let mut guard = self.upload_queue.lock().unwrap();
    1219          511 :         let upload_queue = guard.initialized_mut()?;
    1220              : 
    1221          511 :         self.schedule_deletion_of_unlinked0(upload_queue, layers);
    1222          511 :         self.launch_queued_tasks(upload_queue);
    1223          511 :         Ok(())
    1224          511 :     }
    1225              : 
    1226          517 :     fn schedule_deletion_of_unlinked0(
    1227          517 :         self: &Arc<Self>,
    1228          517 :         upload_queue: &mut UploadQueueInitialized,
    1229          517 :         mut with_metadata: Vec<(LayerName, LayerFileMetadata)>,
    1230          517 :     ) {
    1231          517 :         // Filter out any layers which were not created by this tenant shard.  These are
    1232          517 :         // layers that originate from some ancestor shard after a split, and may still
    1233          517 :         // be referenced by other shards. We are free to delete them locally and remove
    1234          517 :         // them from our index (and would have already done so when we reach this point
    1235          517 :         // in the code), but we may not delete them remotely.
    1236          517 :         with_metadata.retain(|(name, meta)| {
    1237          511 :             let retain = meta.shard.shard_number == self.tenant_shard_id.shard_number
    1238          511 :                 && meta.shard.shard_count == self.tenant_shard_id.shard_count;
    1239          511 :             if !retain {
    1240            0 :                 tracing::debug!(
    1241            0 :                     "Skipping deletion of ancestor-shard layer {name}, from shard {}",
    1242              :                     meta.shard
    1243              :                 );
    1244          511 :             }
    1245          511 :             retain
    1246          517 :         });
    1247              : 
    1248         1028 :         for (name, meta) in &with_metadata {
    1249          511 :             info!(
    1250            0 :                 "scheduling deletion of layer {}{} (shard {})",
    1251            0 :                 name,
    1252            0 :                 meta.generation.get_suffix(),
    1253              :                 meta.shard
    1254              :             );
    1255              :         }
    1256              : 
    1257              :         #[cfg(feature = "testing")]
    1258         1028 :         for (name, meta) in &with_metadata {
    1259          511 :             let gen = meta.generation;
    1260          511 :             match upload_queue.dangling_files.remove(name) {
    1261          441 :                 Some(same) if same == gen => { /* expected */ }
    1262            0 :                 Some(other) => {
    1263            0 :                     tracing::error!("{name} was unlinked with {other:?} but deleted with {gen:?}");
    1264              :                 }
    1265              :                 None => {
    1266           70 :                     tracing::error!("{name} was unlinked but was not dangling");
    1267              :                 }
    1268              :             }
    1269              :         }
    1270              : 
    1271              :         // schedule the actual deletions
    1272          517 :         if with_metadata.is_empty() {
    1273              :             // avoid scheduling the op & bumping the metric
    1274            6 :             return;
    1275          511 :         }
    1276          511 :         let op = UploadOp::Delete(Delete {
    1277          511 :             layers: with_metadata,
    1278          511 :         });
    1279          511 :         self.metric_begin(&op);
    1280          511 :         upload_queue.queued_operations.push_back(op);
    1281          517 :     }
    1282              : 
    1283              :     /// Schedules a compaction update to the remote `index_part.json`.
    1284              :     ///
    1285              :     /// `compacted_from` represent the L0 names which have been `compacted_to` L1 layers.
    1286           76 :     pub(crate) fn schedule_compaction_update(
    1287           76 :         self: &Arc<Self>,
    1288           76 :         compacted_from: &[Layer],
    1289           76 :         compacted_to: &[ResidentLayer],
    1290           76 :     ) -> Result<(), NotInitialized> {
    1291           76 :         let mut guard = self.upload_queue.lock().unwrap();
    1292           76 :         let upload_queue = guard.initialized_mut()?;
    1293              : 
    1294          438 :         for layer in compacted_to {
    1295          362 :             self.schedule_layer_file_upload0(upload_queue, layer.clone());
    1296          362 :         }
    1297              : 
    1298          506 :         let names = compacted_from.iter().map(|x| x.layer_desc().layer_name());
    1299           76 : 
    1300           76 :         self.schedule_unlinking_of_layers_from_index_part0(upload_queue, names);
    1301           76 :         self.launch_queued_tasks(upload_queue);
    1302           76 : 
    1303           76 :         Ok(())
    1304           76 :     }
    1305              : 
    1306              :     /// Wait for all previously scheduled uploads/deletions to complete
    1307         1338 :     pub(crate) async fn wait_completion(self: &Arc<Self>) -> Result<(), WaitCompletionError> {
    1308         1338 :         let receiver = {
    1309         1338 :             let mut guard = self.upload_queue.lock().unwrap();
    1310         1338 :             let upload_queue = guard
    1311         1338 :                 .initialized_mut()
    1312         1338 :                 .map_err(WaitCompletionError::NotInitialized)?;
    1313         1338 :             self.schedule_barrier0(upload_queue)
    1314         1338 :         };
    1315         1338 : 
    1316         1338 :         Self::wait_completion0(receiver).await
    1317         1338 :     }
    1318              : 
    1319         1338 :     async fn wait_completion0(
    1320         1338 :         mut receiver: tokio::sync::watch::Receiver<()>,
    1321         1338 :     ) -> Result<(), WaitCompletionError> {
    1322         1338 :         if receiver.changed().await.is_err() {
    1323            0 :             return Err(WaitCompletionError::UploadQueueShutDownOrStopped);
    1324         1338 :         }
    1325         1338 : 
    1326         1338 :         Ok(())
    1327         1338 :     }
    1328              : 
    1329            6 :     pub(crate) fn schedule_barrier(self: &Arc<Self>) -> anyhow::Result<()> {
    1330            6 :         let mut guard = self.upload_queue.lock().unwrap();
    1331            6 :         let upload_queue = guard.initialized_mut()?;
    1332            6 :         self.schedule_barrier0(upload_queue);
    1333            6 :         Ok(())
    1334            6 :     }
    1335              : 
    1336         1344 :     fn schedule_barrier0(
    1337         1344 :         self: &Arc<Self>,
    1338         1344 :         upload_queue: &mut UploadQueueInitialized,
    1339         1344 :     ) -> tokio::sync::watch::Receiver<()> {
    1340         1344 :         let (sender, receiver) = tokio::sync::watch::channel(());
    1341         1344 :         let barrier_op = UploadOp::Barrier(sender);
    1342         1344 : 
    1343         1344 :         upload_queue.queued_operations.push_back(barrier_op);
    1344         1344 :         // Don't count this kind of operation!
    1345         1344 : 
    1346         1344 :         // Launch the task immediately, if possible
    1347         1344 :         self.launch_queued_tasks(upload_queue);
    1348         1344 : 
    1349         1344 :         receiver
    1350         1344 :     }
    1351              : 
    1352              :     /// Wait for all previously scheduled operations to complete, and then stop.
    1353              :     ///
    1354              :     /// Not cancellation safe
    1355            8 :     pub(crate) async fn shutdown(self: &Arc<Self>) {
    1356            8 :         // On cancellation the queue is left in ackward state of refusing new operations but
    1357            8 :         // proper stop is yet to be called. On cancel the original or some later task must call
    1358            8 :         // `stop` or `shutdown`.
    1359            8 :         let sg = scopeguard::guard((), |_| {
    1360            0 :             tracing::error!("RemoteTimelineClient::shutdown was cancelled; this should not happen, do not make this into an allowed_error")
    1361            8 :         });
    1362              : 
    1363            8 :         let fut = {
    1364            8 :             let mut guard = self.upload_queue.lock().unwrap();
    1365            8 :             let upload_queue = match &mut *guard {
    1366              :                 UploadQueue::Stopped(_) => {
    1367            0 :                     scopeguard::ScopeGuard::into_inner(sg);
    1368            0 :                     return;
    1369              :                 }
    1370              :                 UploadQueue::Uninitialized => {
    1371              :                     // transition into Stopped state
    1372            0 :                     self.stop_impl(&mut guard);
    1373            0 :                     scopeguard::ScopeGuard::into_inner(sg);
    1374            0 :                     return;
    1375              :                 }
    1376            8 :                 UploadQueue::Initialized(ref mut init) => init,
    1377            8 :             };
    1378            8 : 
    1379            8 :             // if the queue is already stuck due to a shutdown operation which was cancelled, then
    1380            8 :             // just don't add more of these as they would never complete.
    1381            8 :             //
    1382            8 :             // TODO: if launch_queued_tasks were to be refactored to accept a &mut UploadQueue
    1383            8 :             // in every place we would not have to jump through this hoop, and this method could be
    1384            8 :             // made cancellable.
    1385            8 :             if !upload_queue.shutting_down {
    1386            8 :                 upload_queue.shutting_down = true;
    1387            8 :                 upload_queue.queued_operations.push_back(UploadOp::Shutdown);
    1388            8 :                 // this operation is not counted similar to Barrier
    1389            8 : 
    1390            8 :                 self.launch_queued_tasks(upload_queue);
    1391            8 :             }
    1392              : 
    1393            8 :             upload_queue.shutdown_ready.clone().acquire_owned()
    1394              :         };
    1395              : 
    1396            8 :         let res = fut.await;
    1397              : 
    1398            8 :         scopeguard::ScopeGuard::into_inner(sg);
    1399            8 : 
    1400            8 :         match res {
    1401            0 :             Ok(_permit) => unreachable!("shutdown_ready should not have been added permits"),
    1402            8 :             Err(_closed) => {
    1403            8 :                 // expected
    1404            8 :             }
    1405            8 :         }
    1406            8 : 
    1407            8 :         self.stop();
    1408            8 :     }
    1409              : 
    1410              :     /// Set the deleted_at field in the remote index file.
    1411              :     ///
    1412              :     /// This fails if the upload queue has not been `stop()`ed.
    1413              :     ///
    1414              :     /// The caller is responsible for calling `stop()` AND for waiting
    1415              :     /// for any ongoing upload tasks to finish after `stop()` has succeeded.
    1416              :     /// Check method [`RemoteTimelineClient::stop`] for details.
    1417            0 :     #[instrument(skip_all)]
    1418              :     pub(crate) async fn persist_index_part_with_deleted_flag(
    1419              :         self: &Arc<Self>,
    1420              :     ) -> Result<(), PersistIndexPartWithDeletedFlagError> {
    1421              :         let index_part_with_deleted_at = {
    1422              :             let mut locked = self.upload_queue.lock().unwrap();
    1423              : 
    1424              :             // We must be in stopped state because otherwise
    1425              :             // we can have inprogress index part upload that can overwrite the file
    1426              :             // with missing is_deleted flag that we going to set below
    1427              :             let stopped = locked.stopped_mut()?;
    1428              : 
    1429              :             match stopped.deleted_at {
    1430              :                 SetDeletedFlagProgress::NotRunning => (), // proceed
    1431              :                 SetDeletedFlagProgress::InProgress(at) => {
    1432              :                     return Err(PersistIndexPartWithDeletedFlagError::AlreadyInProgress(at));
    1433              :                 }
    1434              :                 SetDeletedFlagProgress::Successful(at) => {
    1435              :                     return Err(PersistIndexPartWithDeletedFlagError::AlreadyDeleted(at));
    1436              :                 }
    1437              :             };
    1438              :             let deleted_at = Utc::now().naive_utc();
    1439              :             stopped.deleted_at = SetDeletedFlagProgress::InProgress(deleted_at);
    1440              : 
    1441              :             let mut index_part = stopped.upload_queue_for_deletion.dirty.clone();
    1442              :             index_part.deleted_at = Some(deleted_at);
    1443              :             index_part
    1444              :         };
    1445              : 
    1446            0 :         let undo_deleted_at = scopeguard::guard(Arc::clone(self), |self_clone| {
    1447            0 :             let mut locked = self_clone.upload_queue.lock().unwrap();
    1448            0 :             let stopped = locked
    1449            0 :                 .stopped_mut()
    1450            0 :                 .expect("there's no way out of Stopping, and we checked it's Stopping above");
    1451            0 :             stopped.deleted_at = SetDeletedFlagProgress::NotRunning;
    1452            0 :         });
    1453              : 
    1454              :         pausable_failpoint!("persist_deleted_index_part");
    1455              : 
    1456              :         backoff::retry(
    1457            0 :             || {
    1458            0 :                 upload::upload_index_part(
    1459            0 :                     &self.storage_impl,
    1460            0 :                     &self.tenant_shard_id,
    1461            0 :                     &self.timeline_id,
    1462            0 :                     self.generation,
    1463            0 :                     &index_part_with_deleted_at,
    1464            0 :                     &self.cancel,
    1465            0 :                 )
    1466            0 :             },
    1467            0 :             |_e| false,
    1468              :             1,
    1469              :             // have just a couple of attempts
    1470              :             // when executed as part of timeline deletion this happens in context of api call
    1471              :             // when executed as part of tenant deletion this happens in the background
    1472              :             2,
    1473              :             "persist_index_part_with_deleted_flag",
    1474              :             &self.cancel,
    1475              :         )
    1476              :         .await
    1477            0 :         .ok_or_else(|| anyhow::Error::new(TimeoutOrCancel::Cancel))
    1478            0 :         .and_then(|x| x)?;
    1479              : 
    1480              :         // all good, disarm the guard and mark as success
    1481              :         ScopeGuard::into_inner(undo_deleted_at);
    1482              :         {
    1483              :             let mut locked = self.upload_queue.lock().unwrap();
    1484              : 
    1485              :             let stopped = locked
    1486              :                 .stopped_mut()
    1487              :                 .expect("there's no way out of Stopping, and we checked it's Stopping above");
    1488              :             stopped.deleted_at = SetDeletedFlagProgress::Successful(
    1489              :                 index_part_with_deleted_at
    1490              :                     .deleted_at
    1491              :                     .expect("we set it above"),
    1492              :             );
    1493              :         }
    1494              : 
    1495              :         Ok(())
    1496              :     }
    1497              : 
    1498            0 :     pub(crate) fn is_deleting(&self) -> bool {
    1499            0 :         let mut locked = self.upload_queue.lock().unwrap();
    1500            0 :         locked.stopped_mut().is_ok()
    1501            0 :     }
    1502              : 
    1503            0 :     pub(crate) async fn preserve_initdb_archive(
    1504            0 :         self: &Arc<Self>,
    1505            0 :         tenant_id: &TenantId,
    1506            0 :         timeline_id: &TimelineId,
    1507            0 :         cancel: &CancellationToken,
    1508            0 :     ) -> anyhow::Result<()> {
    1509            0 :         backoff::retry(
    1510            0 :             || async {
    1511            0 :                 upload::preserve_initdb_archive(&self.storage_impl, tenant_id, timeline_id, cancel)
    1512            0 :                     .await
    1513            0 :             },
    1514            0 :             TimeoutOrCancel::caused_by_cancel,
    1515            0 :             FAILED_DOWNLOAD_WARN_THRESHOLD,
    1516            0 :             FAILED_REMOTE_OP_RETRIES,
    1517            0 :             "preserve_initdb_tar_zst",
    1518            0 :             &cancel.clone(),
    1519            0 :         )
    1520            0 :         .await
    1521            0 :         .ok_or_else(|| anyhow::Error::new(TimeoutOrCancel::Cancel))
    1522            0 :         .and_then(|x| x)
    1523            0 :         .context("backing up initdb archive")?;
    1524            0 :         Ok(())
    1525            0 :     }
    1526              : 
    1527              :     /// Uploads the given layer **without** adding it to be part of a future `index_part.json` upload.
    1528              :     ///
    1529              :     /// This is not normally needed.
    1530            0 :     pub(crate) async fn upload_layer_file(
    1531            0 :         self: &Arc<Self>,
    1532            0 :         uploaded: &ResidentLayer,
    1533            0 :         cancel: &CancellationToken,
    1534            0 :     ) -> anyhow::Result<()> {
    1535            0 :         let remote_path = remote_layer_path(
    1536            0 :             &self.tenant_shard_id.tenant_id,
    1537            0 :             &self.timeline_id,
    1538            0 :             uploaded.metadata().shard,
    1539            0 :             &uploaded.layer_desc().layer_name(),
    1540            0 :             uploaded.metadata().generation,
    1541            0 :         );
    1542            0 : 
    1543            0 :         backoff::retry(
    1544            0 :             || async {
    1545            0 :                 upload::upload_timeline_layer(
    1546            0 :                     &self.storage_impl,
    1547            0 :                     uploaded.local_path(),
    1548            0 :                     &remote_path,
    1549            0 :                     uploaded.metadata().file_size,
    1550            0 :                     cancel,
    1551            0 :                 )
    1552            0 :                 .await
    1553            0 :             },
    1554            0 :             TimeoutOrCancel::caused_by_cancel,
    1555            0 :             FAILED_UPLOAD_WARN_THRESHOLD,
    1556            0 :             FAILED_REMOTE_OP_RETRIES,
    1557            0 :             "upload a layer without adding it to latest files",
    1558            0 :             cancel,
    1559            0 :         )
    1560            0 :         .await
    1561            0 :         .ok_or_else(|| anyhow::Error::new(TimeoutOrCancel::Cancel))
    1562            0 :         .and_then(|x| x)
    1563            0 :         .context("upload a layer without adding it to latest files")
    1564            0 :     }
    1565              : 
    1566              :     /// Copies the `adopted` remote existing layer to the remote path of `adopted_as`. The layer is
    1567              :     /// not added to be part of a future `index_part.json` upload.
    1568            0 :     pub(crate) async fn copy_timeline_layer(
    1569            0 :         self: &Arc<Self>,
    1570            0 :         adopted: &Layer,
    1571            0 :         adopted_as: &Layer,
    1572            0 :         cancel: &CancellationToken,
    1573            0 :     ) -> anyhow::Result<()> {
    1574            0 :         let source_remote_path = remote_layer_path(
    1575            0 :             &self.tenant_shard_id.tenant_id,
    1576            0 :             &adopted
    1577            0 :                 .get_timeline_id()
    1578            0 :                 .expect("Source timeline should be alive"),
    1579            0 :             adopted.metadata().shard,
    1580            0 :             &adopted.layer_desc().layer_name(),
    1581            0 :             adopted.metadata().generation,
    1582            0 :         );
    1583            0 : 
    1584            0 :         let target_remote_path = remote_layer_path(
    1585            0 :             &self.tenant_shard_id.tenant_id,
    1586            0 :             &self.timeline_id,
    1587            0 :             adopted_as.metadata().shard,
    1588            0 :             &adopted_as.layer_desc().layer_name(),
    1589            0 :             adopted_as.metadata().generation,
    1590            0 :         );
    1591            0 : 
    1592            0 :         backoff::retry(
    1593            0 :             || async {
    1594            0 :                 upload::copy_timeline_layer(
    1595            0 :                     &self.storage_impl,
    1596            0 :                     &source_remote_path,
    1597            0 :                     &target_remote_path,
    1598            0 :                     cancel,
    1599            0 :                 )
    1600            0 :                 .await
    1601            0 :             },
    1602            0 :             TimeoutOrCancel::caused_by_cancel,
    1603            0 :             FAILED_UPLOAD_WARN_THRESHOLD,
    1604            0 :             FAILED_REMOTE_OP_RETRIES,
    1605            0 :             "copy timeline layer",
    1606            0 :             cancel,
    1607            0 :         )
    1608            0 :         .await
    1609            0 :         .ok_or_else(|| anyhow::Error::new(TimeoutOrCancel::Cancel))
    1610            0 :         .and_then(|x| x)
    1611            0 :         .context("remote copy timeline layer")
    1612            0 :     }
    1613              : 
    1614            0 :     async fn flush_deletion_queue(&self) -> Result<(), DeletionQueueError> {
    1615            0 :         match tokio::time::timeout(
    1616            0 :             DELETION_QUEUE_FLUSH_TIMEOUT,
    1617            0 :             self.deletion_queue_client.flush_immediate(),
    1618            0 :         )
    1619            0 :         .await
    1620              :         {
    1621            0 :             Ok(result) => result,
    1622            0 :             Err(_timeout) => {
    1623            0 :                 // Flushing remote deletions is not mandatory: we flush here to make the system easier to test, and
    1624            0 :                 // to ensure that _usually_ objects are really gone after a DELETE is acked.  However, in case of deletion
    1625            0 :                 // queue issues (https://github.com/neondatabase/neon/issues/6440), we don't want to wait indefinitely here.
    1626            0 :                 tracing::warn!(
    1627            0 :                     "Timed out waiting for deletion queue flush, acking deletion anyway"
    1628              :                 );
    1629            0 :                 Ok(())
    1630              :             }
    1631              :         }
    1632            0 :     }
    1633              : 
    1634              :     /// Prerequisites: UploadQueue should be in stopped state and deleted_at should be successfuly set.
    1635              :     /// The function deletes layer files one by one, then lists the prefix to see if we leaked something
    1636              :     /// deletes leaked files if any and proceeds with deletion of index file at the end.
    1637            0 :     pub(crate) async fn delete_all(self: &Arc<Self>) -> Result<(), DeleteTimelineError> {
    1638            0 :         debug_assert_current_span_has_tenant_and_timeline_id();
    1639              : 
    1640            0 :         let layers: Vec<RemotePath> = {
    1641            0 :             let mut locked = self.upload_queue.lock().unwrap();
    1642            0 :             let stopped = locked.stopped_mut().map_err(DeleteTimelineError::Other)?;
    1643              : 
    1644            0 :             if !matches!(stopped.deleted_at, SetDeletedFlagProgress::Successful(_)) {
    1645            0 :                 return Err(DeleteTimelineError::Other(anyhow::anyhow!(
    1646            0 :                     "deleted_at is not set"
    1647            0 :                 )));
    1648            0 :             }
    1649            0 : 
    1650            0 :             debug_assert!(stopped.upload_queue_for_deletion.no_pending_work());
    1651              : 
    1652            0 :             stopped
    1653            0 :                 .upload_queue_for_deletion
    1654            0 :                 .dirty
    1655            0 :                 .layer_metadata
    1656            0 :                 .drain()
    1657            0 :                 .filter(|(_file_name, meta)| {
    1658            0 :                     // Filter out layers that belonged to an ancestor shard.  Since we are deleting the whole timeline from
    1659            0 :                     // all shards anyway, we _could_ delete these, but
    1660            0 :                     // - it creates a potential race if other shards are still
    1661            0 :                     //   using the layers while this shard deletes them.
    1662            0 :                     // - it means that if we rolled back the shard split, the ancestor shards would be in a state where
    1663            0 :                     //   these timelines are present but corrupt (their index exists but some layers don't)
    1664            0 :                     //
    1665            0 :                     // These layers will eventually be cleaned up by the scrubber when it does physical GC.
    1666            0 :                     meta.shard.shard_number == self.tenant_shard_id.shard_number
    1667            0 :                         && meta.shard.shard_count == self.tenant_shard_id.shard_count
    1668            0 :                 })
    1669            0 :                 .map(|(file_name, meta)| {
    1670            0 :                     remote_layer_path(
    1671            0 :                         &self.tenant_shard_id.tenant_id,
    1672            0 :                         &self.timeline_id,
    1673            0 :                         meta.shard,
    1674            0 :                         &file_name,
    1675            0 :                         meta.generation,
    1676            0 :                     )
    1677            0 :                 })
    1678            0 :                 .collect()
    1679            0 :         };
    1680            0 : 
    1681            0 :         let layer_deletion_count = layers.len();
    1682            0 :         self.deletion_queue_client
    1683            0 :             .push_immediate(layers)
    1684            0 :             .await
    1685            0 :             .map_err(|_| DeleteTimelineError::Cancelled)?;
    1686              : 
    1687              :         // Delete the initdb.tar.zst, which is not always present, but deletion attempts of
    1688              :         // inexistant objects are not considered errors.
    1689            0 :         let initdb_path =
    1690            0 :             remote_initdb_archive_path(&self.tenant_shard_id.tenant_id, &self.timeline_id);
    1691            0 :         self.deletion_queue_client
    1692            0 :             .push_immediate(vec![initdb_path])
    1693            0 :             .await
    1694            0 :             .map_err(|_| DeleteTimelineError::Cancelled)?;
    1695              : 
    1696              :         // Do not delete index part yet, it is needed for possible retry. If we remove it first
    1697              :         // and retry will arrive to different pageserver there wont be any traces of it on remote storage
    1698            0 :         let timeline_storage_path = remote_timeline_path(&self.tenant_shard_id, &self.timeline_id);
    1699            0 : 
    1700            0 :         // Execute all pending deletions, so that when we proceed to do a listing below, we aren't
    1701            0 :         // taking the burden of listing all the layers that we already know we should delete.
    1702            0 :         self.flush_deletion_queue()
    1703            0 :             .await
    1704            0 :             .map_err(|_| DeleteTimelineError::Cancelled)?;
    1705              : 
    1706            0 :         let cancel = shutdown_token();
    1707              : 
    1708            0 :         let remaining = download_retry(
    1709            0 :             || async {
    1710            0 :                 self.storage_impl
    1711            0 :                     .list(
    1712            0 :                         Some(&timeline_storage_path),
    1713            0 :                         ListingMode::NoDelimiter,
    1714            0 :                         None,
    1715            0 :                         &cancel,
    1716            0 :                     )
    1717            0 :                     .await
    1718            0 :             },
    1719            0 :             "list remaining files",
    1720            0 :             &cancel,
    1721            0 :         )
    1722            0 :         .await
    1723            0 :         .context("list files remaining files")?
    1724              :         .keys;
    1725              : 
    1726              :         // We will delete the current index_part object last, since it acts as a deletion
    1727              :         // marker via its deleted_at attribute
    1728            0 :         let latest_index = remaining
    1729            0 :             .iter()
    1730            0 :             .filter(|o| {
    1731            0 :                 o.key
    1732            0 :                     .object_name()
    1733            0 :                     .map(|n| n.starts_with(IndexPart::FILE_NAME))
    1734            0 :                     .unwrap_or(false)
    1735            0 :             })
    1736            0 :             .filter_map(|o| parse_remote_index_path(o.key.clone()).map(|gen| (o.key.clone(), gen)))
    1737            0 :             .max_by_key(|i| i.1)
    1738            0 :             .map(|i| i.0.clone())
    1739            0 :             .unwrap_or(
    1740            0 :                 // No generation-suffixed indices, assume we are dealing with
    1741            0 :                 // a legacy index.
    1742            0 :                 remote_index_path(&self.tenant_shard_id, &self.timeline_id, Generation::none()),
    1743            0 :             );
    1744            0 : 
    1745            0 :         let remaining_layers: Vec<RemotePath> = remaining
    1746            0 :             .into_iter()
    1747            0 :             .filter_map(|o| {
    1748            0 :                 if o.key == latest_index || o.key.object_name() == Some(INITDB_PRESERVED_PATH) {
    1749            0 :                     None
    1750              :                 } else {
    1751            0 :                     Some(o.key)
    1752              :                 }
    1753            0 :             })
    1754            0 :             .inspect(|path| {
    1755            0 :                 if let Some(name) = path.object_name() {
    1756            0 :                     info!(%name, "deleting a file not referenced from index_part.json");
    1757              :                 } else {
    1758            0 :                     warn!(%path, "deleting a nameless or non-utf8 object not referenced from index_part.json");
    1759              :                 }
    1760            0 :             })
    1761            0 :             .collect();
    1762            0 : 
    1763            0 :         let not_referenced_count = remaining_layers.len();
    1764            0 :         if !remaining_layers.is_empty() {
    1765            0 :             self.deletion_queue_client
    1766            0 :                 .push_immediate(remaining_layers)
    1767            0 :                 .await
    1768            0 :                 .map_err(|_| DeleteTimelineError::Cancelled)?;
    1769            0 :         }
    1770              : 
    1771            0 :         fail::fail_point!("timeline-delete-before-index-delete", |_| {
    1772            0 :             Err(DeleteTimelineError::Other(anyhow::anyhow!(
    1773            0 :                 "failpoint: timeline-delete-before-index-delete"
    1774            0 :             )))?
    1775            0 :         });
    1776              : 
    1777            0 :         debug!("enqueuing index part deletion");
    1778            0 :         self.deletion_queue_client
    1779            0 :             .push_immediate([latest_index].to_vec())
    1780            0 :             .await
    1781            0 :             .map_err(|_| DeleteTimelineError::Cancelled)?;
    1782              : 
    1783              :         // Timeline deletion is rare and we have probably emitted a reasonably number of objects: wait
    1784              :         // for a flush to a persistent deletion list so that we may be sure deletion will occur.
    1785            0 :         self.flush_deletion_queue()
    1786            0 :             .await
    1787            0 :             .map_err(|_| DeleteTimelineError::Cancelled)?;
    1788              : 
    1789            0 :         fail::fail_point!("timeline-delete-after-index-delete", |_| {
    1790            0 :             Err(DeleteTimelineError::Other(anyhow::anyhow!(
    1791            0 :                 "failpoint: timeline-delete-after-index-delete"
    1792            0 :             )))?
    1793            0 :         });
    1794              : 
    1795            0 :         info!(prefix=%timeline_storage_path, referenced=layer_deletion_count, not_referenced=%not_referenced_count, "done deleting in timeline prefix, including index_part.json");
    1796              : 
    1797            0 :         Ok(())
    1798            0 :     }
    1799              : 
    1800              :     ///
    1801              :     /// Pick next tasks from the queue, and start as many of them as possible without violating
    1802              :     /// the ordering constraints.
    1803              :     ///
    1804              :     /// The caller needs to already hold the `upload_queue` lock.
    1805         7624 :     fn launch_queued_tasks(self: &Arc<Self>, upload_queue: &mut UploadQueueInitialized) {
    1806        12250 :         while let Some(next_op) = upload_queue.queued_operations.front() {
    1807              :             // Can we run this task now?
    1808         8872 :             let can_run_now = match next_op {
    1809              :                 UploadOp::UploadLayer(..) => {
    1810              :                     // Can always be scheduled.
    1811         1549 :                     true
    1812              :                 }
    1813              :                 UploadOp::UploadMetadata { .. } => {
    1814              :                     // These can only be performed after all the preceding operations
    1815              :                     // have finished.
    1816         4493 :                     upload_queue.inprogress_tasks.is_empty()
    1817              :                 }
    1818              :                 UploadOp::Delete(..) => {
    1819              :                     // Wait for preceding uploads to finish. Concurrent deletions are OK, though.
    1820          326 :                     upload_queue.num_inprogress_deletions == upload_queue.inprogress_tasks.len()
    1821              :                 }
    1822              : 
    1823              :                 UploadOp::Barrier(_) | UploadOp::Shutdown => {
    1824         2504 :                     upload_queue.inprogress_tasks.is_empty()
    1825              :                 }
    1826              :             };
    1827              : 
    1828              :             // If we cannot launch this task, don't look any further.
    1829              :             //
    1830              :             // In some cases, we could let some non-frontmost tasks to "jump the queue" and launch
    1831              :             // them now, but we don't try to do that currently.  For example, if the frontmost task
    1832              :             // is an index-file upload that cannot proceed until preceding uploads have finished, we
    1833              :             // could still start layer uploads that were scheduled later.
    1834         8872 :             if !can_run_now {
    1835         4238 :                 break;
    1836         4634 :             }
    1837         4634 : 
    1838         4634 :             if let UploadOp::Shutdown = next_op {
    1839              :                 // leave the op in the queue but do not start more tasks; it will be dropped when
    1840              :                 // the stop is called.
    1841            8 :                 upload_queue.shutdown_ready.close();
    1842            8 :                 break;
    1843         4626 :             }
    1844         4626 : 
    1845         4626 :             // We can launch this task. Remove it from the queue first.
    1846         4626 :             let mut next_op = upload_queue.queued_operations.pop_front().unwrap();
    1847         4626 : 
    1848         4626 :             debug!("starting op: {}", next_op);
    1849              : 
    1850              :             // Update the counters and prepare
    1851         4626 :             match &mut next_op {
    1852         1549 :                 UploadOp::UploadLayer(layer, meta, mode) => {
    1853         1549 :                     if upload_queue
    1854         1549 :                         .recently_deleted
    1855         1549 :                         .remove(&(layer.layer_desc().layer_name().clone(), meta.generation))
    1856            0 :                     {
    1857            0 :                         *mode = Some(OpType::FlushDeletion);
    1858            0 :                     } else {
    1859         1549 :                         *mode = Some(OpType::MayReorder)
    1860              :                     }
    1861         1549 :                     upload_queue.num_inprogress_layer_uploads += 1;
    1862              :                 }
    1863         1478 :                 UploadOp::UploadMetadata { .. } => {
    1864         1478 :                     upload_queue.num_inprogress_metadata_uploads += 1;
    1865         1478 :                 }
    1866          255 :                 UploadOp::Delete(Delete { layers }) => {
    1867          510 :                     for (name, meta) in layers {
    1868          255 :                         upload_queue
    1869          255 :                             .recently_deleted
    1870          255 :                             .insert((name.clone(), meta.generation));
    1871          255 :                     }
    1872          255 :                     upload_queue.num_inprogress_deletions += 1;
    1873              :                 }
    1874         1344 :                 UploadOp::Barrier(sender) => {
    1875         1344 :                     sender.send_replace(());
    1876         1344 :                     continue;
    1877              :                 }
    1878            0 :                 UploadOp::Shutdown => unreachable!("shutdown is intentionally never popped off"),
    1879              :             };
    1880              : 
    1881              :             // Assign unique ID to this task
    1882         3282 :             upload_queue.task_counter += 1;
    1883         3282 :             let upload_task_id = upload_queue.task_counter;
    1884         3282 : 
    1885         3282 :             // Add it to the in-progress map
    1886         3282 :             let task = Arc::new(UploadTask {
    1887         3282 :                 task_id: upload_task_id,
    1888         3282 :                 op: next_op,
    1889         3282 :                 retries: AtomicU32::new(0),
    1890         3282 :             });
    1891         3282 :             upload_queue
    1892         3282 :                 .inprogress_tasks
    1893         3282 :                 .insert(task.task_id, Arc::clone(&task));
    1894         3282 : 
    1895         3282 :             // Spawn task to perform the task
    1896         3282 :             let self_rc = Arc::clone(self);
    1897         3282 :             let tenant_shard_id = self.tenant_shard_id;
    1898         3282 :             let timeline_id = self.timeline_id;
    1899         3282 :             task_mgr::spawn(
    1900         3282 :                 &self.runtime,
    1901         3282 :                 TaskKind::RemoteUploadTask,
    1902         3282 :                 self.tenant_shard_id,
    1903         3282 :                 Some(self.timeline_id),
    1904         3282 :                 "remote upload",
    1905         3174 :                 async move {
    1906         3174 :                     self_rc.perform_upload_task(task).await;
    1907         2967 :                     Ok(())
    1908         2967 :                 }
    1909         3282 :                 .instrument(info_span!(parent: None, "remote_upload", tenant_id=%tenant_shard_id.tenant_id, shard_id=%tenant_shard_id.shard_slug(), %timeline_id, %upload_task_id)),
    1910              :             );
    1911              : 
    1912              :             // Loop back to process next task
    1913              :         }
    1914         7624 :     }
    1915              : 
    1916              :     ///
    1917              :     /// Perform an upload task.
    1918              :     ///
    1919              :     /// The task is in the `inprogress_tasks` list. This function will try to
    1920              :     /// execute it, retrying forever. On successful completion, the task is
    1921              :     /// removed it from the `inprogress_tasks` list, and any next task(s) in the
    1922              :     /// queue that were waiting by the completion are launched.
    1923              :     ///
    1924              :     /// The task can be shut down, however. That leads to stopping the whole
    1925              :     /// queue.
    1926              :     ///
    1927         3174 :     async fn perform_upload_task(self: &Arc<Self>, task: Arc<UploadTask>) {
    1928         3174 :         let cancel = shutdown_token();
    1929              :         // Loop to retry until it completes.
    1930              :         loop {
    1931              :             // If we're requested to shut down, close up shop and exit.
    1932              :             //
    1933              :             // Note: We only check for the shutdown requests between retries, so
    1934              :             // if a shutdown request arrives while we're busy uploading, in the
    1935              :             // upload::upload:*() call below, we will wait not exit until it has
    1936              :             // finished. We probably could cancel the upload by simply dropping
    1937              :             // the Future, but we're not 100% sure if the remote storage library
    1938              :             // is cancellation safe, so we don't dare to do that. Hopefully, the
    1939              :             // upload finishes or times out soon enough.
    1940         3174 :             if cancel.is_cancelled() {
    1941            0 :                 info!("upload task cancelled by shutdown request");
    1942            0 :                 self.stop();
    1943            0 :                 return;
    1944         3174 :             }
    1945         3174 : 
    1946         3174 :             // Assert that we don't modify a layer that's referenced by the current index.
    1947         3174 :             if cfg!(debug_assertions) {
    1948         3174 :                 let modified = match &task.op {
    1949         1456 :                     UploadOp::UploadLayer(layer, layer_metadata, _) => {
    1950         1456 :                         vec![(layer.layer_desc().layer_name(), layer_metadata)]
    1951              :                     }
    1952          255 :                     UploadOp::Delete(delete) => {
    1953          255 :                         delete.layers.iter().map(|(n, m)| (n.clone(), m)).collect()
    1954              :                     }
    1955              :                     // These don't modify layers.
    1956         1463 :                     UploadOp::UploadMetadata { .. } => Vec::new(),
    1957            0 :                     UploadOp::Barrier(_) => Vec::new(),
    1958            0 :                     UploadOp::Shutdown => Vec::new(),
    1959              :                 };
    1960         3174 :                 if let Ok(queue) = self.upload_queue.lock().unwrap().initialized_mut() {
    1961         4881 :                     for (ref name, metadata) in modified {
    1962         1711 :                         debug_assert!(
    1963         1711 :                             !queue.clean.0.references(name, metadata),
    1964            4 :                             "layer {name} modified while referenced by index",
    1965              :                         );
    1966              :                     }
    1967            0 :                 }
    1968            0 :             }
    1969              : 
    1970         3170 :             let upload_result: anyhow::Result<()> = match &task.op {
    1971         1456 :                 UploadOp::UploadLayer(ref layer, ref layer_metadata, mode) => {
    1972         1456 :                     if let Some(OpType::FlushDeletion) = mode {
    1973            0 :                         if self.config.read().unwrap().block_deletions {
    1974              :                             // Of course, this is not efficient... but usually the queue should be empty.
    1975            0 :                             let mut queue_locked = self.upload_queue.lock().unwrap();
    1976            0 :                             let mut detected = false;
    1977            0 :                             if let Ok(queue) = queue_locked.initialized_mut() {
    1978            0 :                                 for list in queue.blocked_deletions.iter_mut() {
    1979            0 :                                     list.layers.retain(|(name, meta)| {
    1980            0 :                                         if name == &layer.layer_desc().layer_name()
    1981            0 :                                             && meta.generation == layer_metadata.generation
    1982              :                                         {
    1983            0 :                                             detected = true;
    1984            0 :                                             // remove the layer from deletion queue
    1985            0 :                                             false
    1986              :                                         } else {
    1987              :                                             // keep the layer
    1988            0 :                                             true
    1989              :                                         }
    1990            0 :                                     });
    1991            0 :                                 }
    1992            0 :                             }
    1993            0 :                             if detected {
    1994            0 :                                 info!(
    1995            0 :                                     "cancelled blocked deletion of layer {} at gen {:?}",
    1996            0 :                                     layer.layer_desc().layer_name(),
    1997              :                                     layer_metadata.generation
    1998              :                                 );
    1999            0 :                             }
    2000              :                         } else {
    2001              :                             // TODO: we did not guarantee that upload task starts after deletion task, so there could be possibly race conditions
    2002              :                             // that we still get the layer deleted. But this only happens if someone creates a layer immediately after it's deleted,
    2003              :                             // which is not possible in the current system.
    2004            0 :                             info!(
    2005            0 :                                 "waiting for deletion queue flush to complete before uploading layer {} at gen {:?}",
    2006            0 :                                 layer.layer_desc().layer_name(),
    2007              :                                 layer_metadata.generation
    2008              :                             );
    2009              :                             {
    2010              :                                 // We are going to flush, we can clean up the recently deleted list.
    2011            0 :                                 let mut queue_locked = self.upload_queue.lock().unwrap();
    2012            0 :                                 if let Ok(queue) = queue_locked.initialized_mut() {
    2013            0 :                                     queue.recently_deleted.clear();
    2014            0 :                                 }
    2015              :                             }
    2016            0 :                             if let Err(e) = self.deletion_queue_client.flush_execute().await {
    2017            0 :                                 warn!(
    2018            0 :                                     "failed to flush the deletion queue before uploading layer {} at gen {:?}, still proceeding to upload: {e:#} ",
    2019            0 :                                     layer.layer_desc().layer_name(),
    2020              :                                     layer_metadata.generation
    2021              :                                 );
    2022              :                             } else {
    2023            0 :                                 info!(
    2024            0 :                                     "done flushing deletion queue before uploading layer {} at gen {:?}",
    2025            0 :                                     layer.layer_desc().layer_name(),
    2026              :                                     layer_metadata.generation
    2027              :                                 );
    2028              :                             }
    2029              :                         }
    2030         1456 :                     }
    2031         1456 :                     let local_path = layer.local_path();
    2032         1456 : 
    2033         1456 :                     // We should only be uploading layers created by this `Tenant`'s lifetime, so
    2034         1456 :                     // the metadata in the upload should always match our current generation.
    2035         1456 :                     assert_eq!(layer_metadata.generation, self.generation);
    2036              : 
    2037         1456 :                     let remote_path = remote_layer_path(
    2038         1456 :                         &self.tenant_shard_id.tenant_id,
    2039         1456 :                         &self.timeline_id,
    2040         1456 :                         layer_metadata.shard,
    2041         1456 :                         &layer.layer_desc().layer_name(),
    2042         1456 :                         layer_metadata.generation,
    2043         1456 :                     );
    2044         1456 : 
    2045         1456 :                     upload::upload_timeline_layer(
    2046         1456 :                         &self.storage_impl,
    2047         1456 :                         local_path,
    2048         1456 :                         &remote_path,
    2049         1456 :                         layer_metadata.file_size,
    2050         1456 :                         &self.cancel,
    2051         1456 :                     )
    2052         1456 :                     .measure_remote_op(
    2053         1456 :                         RemoteOpFileKind::Layer,
    2054         1456 :                         RemoteOpKind::Upload,
    2055         1456 :                         Arc::clone(&self.metrics),
    2056         1456 :                     )
    2057         1456 :                     .await
    2058              :                 }
    2059         1463 :                 UploadOp::UploadMetadata { ref uploaded } => {
    2060         1463 :                     let res = upload::upload_index_part(
    2061         1463 :                         &self.storage_impl,
    2062         1463 :                         &self.tenant_shard_id,
    2063         1463 :                         &self.timeline_id,
    2064         1463 :                         self.generation,
    2065         1463 :                         uploaded,
    2066         1463 :                         &self.cancel,
    2067         1463 :                     )
    2068         1463 :                     .measure_remote_op(
    2069         1463 :                         RemoteOpFileKind::Index,
    2070         1463 :                         RemoteOpKind::Upload,
    2071         1463 :                         Arc::clone(&self.metrics),
    2072         1463 :                     )
    2073         1463 :                     .await;
    2074         1456 :                     if res.is_ok() {
    2075         1456 :                         self.update_remote_physical_size_gauge(Some(uploaded));
    2076         1456 :                         let mention_having_future_layers = if cfg!(feature = "testing") {
    2077         1456 :                             uploaded
    2078         1456 :                                 .layer_metadata
    2079         1456 :                                 .keys()
    2080        17544 :                                 .any(|x| x.is_in_future(uploaded.metadata.disk_consistent_lsn()))
    2081              :                         } else {
    2082            0 :                             false
    2083              :                         };
    2084         1456 :                         if mention_having_future_layers {
    2085              :                             // find rationale near crate::tenant::timeline::init::cleanup_future_layer
    2086           43 :                             tracing::info!(
    2087            0 :                                 disk_consistent_lsn = %uploaded.metadata.disk_consistent_lsn(),
    2088            0 :                                 "uploaded an index_part.json with future layers -- this is ok! if shutdown now, expect future layer cleanup"
    2089              :                             );
    2090         1413 :                         }
    2091            0 :                     }
    2092         1456 :                     res
    2093              :                 }
    2094          251 :                 UploadOp::Delete(delete) => {
    2095          251 :                     if self.config.read().unwrap().block_deletions {
    2096            0 :                         let mut queue_locked = self.upload_queue.lock().unwrap();
    2097            0 :                         if let Ok(queue) = queue_locked.initialized_mut() {
    2098            0 :                             queue.blocked_deletions.push(delete.clone());
    2099            0 :                         }
    2100            0 :                         Ok(())
    2101              :                     } else {
    2102          251 :                         pausable_failpoint!("before-delete-layer-pausable");
    2103          251 :                         self.deletion_queue_client
    2104          251 :                             .push_layers(
    2105          251 :                                 self.tenant_shard_id,
    2106          251 :                                 self.timeline_id,
    2107          251 :                                 self.generation,
    2108          251 :                                 delete.layers.clone(),
    2109          251 :                             )
    2110          251 :                             .await
    2111          251 :                             .map_err(|e| anyhow::anyhow!(e))
    2112              :                     }
    2113              :                 }
    2114            0 :                 unexpected @ UploadOp::Barrier(_) | unexpected @ UploadOp::Shutdown => {
    2115              :                     // unreachable. Barrier operations are handled synchronously in
    2116              :                     // launch_queued_tasks
    2117            0 :                     warn!("unexpected {unexpected:?} operation in perform_upload_task");
    2118            0 :                     break;
    2119              :                 }
    2120              :             };
    2121              : 
    2122            0 :             match upload_result {
    2123              :                 Ok(()) => {
    2124         2967 :                     break;
    2125              :                 }
    2126            0 :                 Err(e) if TimeoutOrCancel::caused_by_cancel(&e) => {
    2127            0 :                     // loop around to do the proper stopping
    2128            0 :                     continue;
    2129              :                 }
    2130            0 :                 Err(e) => {
    2131            0 :                     let retries = task.retries.fetch_add(1, Ordering::SeqCst);
    2132            0 : 
    2133            0 :                     // Uploads can fail due to rate limits (IAM, S3), spurious network problems,
    2134            0 :                     // or other external reasons. Such issues are relatively regular, so log them
    2135            0 :                     // at info level at first, and only WARN if the operation fails repeatedly.
    2136            0 :                     //
    2137            0 :                     // (See similar logic for downloads in `download::download_retry`)
    2138            0 :                     if retries < FAILED_UPLOAD_WARN_THRESHOLD {
    2139            0 :                         info!(
    2140            0 :                             "failed to perform remote task {}, will retry (attempt {}): {:#}",
    2141            0 :                             task.op, retries, e
    2142              :                         );
    2143              :                     } else {
    2144            0 :                         warn!(
    2145            0 :                             "failed to perform remote task {}, will retry (attempt {}): {:?}",
    2146            0 :                             task.op, retries, e
    2147              :                         );
    2148              :                     }
    2149              : 
    2150              :                     // sleep until it's time to retry, or we're cancelled
    2151            0 :                     exponential_backoff(
    2152            0 :                         retries,
    2153            0 :                         DEFAULT_BASE_BACKOFF_SECONDS,
    2154            0 :                         DEFAULT_MAX_BACKOFF_SECONDS,
    2155            0 :                         &cancel,
    2156            0 :                     )
    2157            0 :                     .await;
    2158              :                 }
    2159              :             }
    2160              :         }
    2161              : 
    2162         2967 :         let retries = task.retries.load(Ordering::SeqCst);
    2163         2967 :         if retries > 0 {
    2164            0 :             info!(
    2165            0 :                 "remote task {} completed successfully after {} retries",
    2166            0 :                 task.op, retries
    2167              :             );
    2168              :         } else {
    2169         2967 :             debug!("remote task {} completed successfully", task.op);
    2170              :         }
    2171              : 
    2172              :         // The task has completed successfully. Remove it from the in-progress list.
    2173         2967 :         let lsn_update = {
    2174         2967 :             let mut upload_queue_guard = self.upload_queue.lock().unwrap();
    2175         2967 :             let upload_queue = match upload_queue_guard.deref_mut() {
    2176            0 :                 UploadQueue::Uninitialized => panic!("callers are responsible for ensuring this is only called on an initialized queue"),
    2177            0 :                 UploadQueue::Stopped(_stopped) => {
    2178            0 :                     None
    2179              :                 },
    2180         2967 :                 UploadQueue::Initialized(qi) => { Some(qi) }
    2181              :             };
    2182              : 
    2183         2967 :             let upload_queue = match upload_queue {
    2184         2967 :                 Some(upload_queue) => upload_queue,
    2185              :                 None => {
    2186            0 :                     info!("another concurrent task already stopped the queue");
    2187            0 :                     return;
    2188              :                 }
    2189              :             };
    2190              : 
    2191         2967 :             upload_queue.inprogress_tasks.remove(&task.task_id);
    2192              : 
    2193         2967 :             let lsn_update = match task.op {
    2194              :                 UploadOp::UploadLayer(_, _, _) => {
    2195         1260 :                     upload_queue.num_inprogress_layer_uploads -= 1;
    2196         1260 :                     None
    2197              :                 }
    2198         1456 :                 UploadOp::UploadMetadata { ref uploaded } => {
    2199         1456 :                     upload_queue.num_inprogress_metadata_uploads -= 1;
    2200         1456 : 
    2201         1456 :                     // the task id is reused as a monotonicity check for storing the "clean"
    2202         1456 :                     // IndexPart.
    2203         1456 :                     let last_updater = upload_queue.clean.1;
    2204         1456 :                     let is_later = last_updater.is_some_and(|task_id| task_id < task.task_id);
    2205         1456 :                     let monotone = is_later || last_updater.is_none();
    2206              : 
    2207         1456 :                     assert!(monotone, "no two index uploads should be completing at the same time, prev={last_updater:?}, task.task_id={}", task.task_id);
    2208              : 
    2209              :                     // not taking ownership is wasteful
    2210         1456 :                     upload_queue.clean.0.clone_from(uploaded);
    2211         1456 :                     upload_queue.clean.1 = Some(task.task_id);
    2212         1456 : 
    2213         1456 :                     let lsn = upload_queue.clean.0.metadata.disk_consistent_lsn();
    2214         1456 :                     self.metrics
    2215         1456 :                         .projected_remote_consistent_lsn_gauge
    2216         1456 :                         .set(lsn.0);
    2217         1456 : 
    2218         1456 :                     if self.generation.is_none() {
    2219              :                         // Legacy mode: skip validating generation
    2220            0 :                         upload_queue.visible_remote_consistent_lsn.store(lsn);
    2221            0 :                         None
    2222         1456 :                     } else if self
    2223         1456 :                         .config
    2224         1456 :                         .read()
    2225         1456 :                         .unwrap()
    2226         1456 :                         .process_remote_consistent_lsn_updates
    2227              :                     {
    2228         1456 :                         Some((lsn, upload_queue.visible_remote_consistent_lsn.clone()))
    2229              :                     } else {
    2230              :                         // Our config disables remote_consistent_lsn updates: drop it.
    2231            0 :                         None
    2232              :                     }
    2233              :                 }
    2234              :                 UploadOp::Delete(_) => {
    2235          251 :                     upload_queue.num_inprogress_deletions -= 1;
    2236          251 :                     None
    2237              :                 }
    2238            0 :                 UploadOp::Barrier(..) | UploadOp::Shutdown => unreachable!(),
    2239              :             };
    2240              : 
    2241              :             // Launch any queued tasks that were unblocked by this one.
    2242         2967 :             self.launch_queued_tasks(upload_queue);
    2243         2967 :             lsn_update
    2244              :         };
    2245              : 
    2246         2967 :         if let Some((lsn, slot)) = lsn_update {
    2247              :             // Updates to the remote_consistent_lsn we advertise to pageservers
    2248              :             // are all routed through the DeletionQueue, to enforce important
    2249              :             // data safety guarantees (see docs/rfcs/025-generation-numbers.md)
    2250         1456 :             self.deletion_queue_client
    2251         1456 :                 .update_remote_consistent_lsn(
    2252         1456 :                     self.tenant_shard_id,
    2253         1456 :                     self.timeline_id,
    2254         1456 :                     self.generation,
    2255         1456 :                     lsn,
    2256         1456 :                     slot,
    2257         1456 :                 )
    2258         1456 :                 .await;
    2259         1511 :         }
    2260              : 
    2261         2967 :         self.metric_end(&task.op);
    2262         2967 :     }
    2263              : 
    2264         6556 :     fn metric_impl(
    2265         6556 :         &self,
    2266         6556 :         op: &UploadOp,
    2267         6556 :     ) -> Option<(
    2268         6556 :         RemoteOpFileKind,
    2269         6556 :         RemoteOpKind,
    2270         6556 :         RemoteTimelineClientMetricsCallTrackSize,
    2271         6556 :     )> {
    2272              :         use RemoteTimelineClientMetricsCallTrackSize::DontTrackSize;
    2273         6556 :         let res = match op {
    2274         2816 :             UploadOp::UploadLayer(_, m, _) => (
    2275         2816 :                 RemoteOpFileKind::Layer,
    2276         2816 :                 RemoteOpKind::Upload,
    2277         2816 :                 RemoteTimelineClientMetricsCallTrackSize::Bytes(m.file_size),
    2278         2816 :             ),
    2279         2970 :             UploadOp::UploadMetadata { .. } => (
    2280         2970 :                 RemoteOpFileKind::Index,
    2281         2970 :                 RemoteOpKind::Upload,
    2282         2970 :                 DontTrackSize {
    2283         2970 :                     reason: "metadata uploads are tiny",
    2284         2970 :                 },
    2285         2970 :             ),
    2286          762 :             UploadOp::Delete(_delete) => (
    2287          762 :                 RemoteOpFileKind::Layer,
    2288          762 :                 RemoteOpKind::Delete,
    2289          762 :                 DontTrackSize {
    2290          762 :                     reason: "should we track deletes? positive or negative sign?",
    2291          762 :                 },
    2292          762 :             ),
    2293              :             UploadOp::Barrier(..) | UploadOp::Shutdown => {
    2294              :                 // we do not account these
    2295            8 :                 return None;
    2296              :             }
    2297              :         };
    2298         6548 :         Some(res)
    2299         6556 :     }
    2300              : 
    2301         3581 :     fn metric_begin(&self, op: &UploadOp) {
    2302         3581 :         let (file_kind, op_kind, track_bytes) = match self.metric_impl(op) {
    2303         3581 :             Some(x) => x,
    2304            0 :             None => return,
    2305              :         };
    2306         3581 :         let guard = self.metrics.call_begin(&file_kind, &op_kind, track_bytes);
    2307         3581 :         guard.will_decrement_manually(); // in metric_end(), see right below
    2308         3581 :     }
    2309              : 
    2310         2975 :     fn metric_end(&self, op: &UploadOp) {
    2311         2975 :         let (file_kind, op_kind, track_bytes) = match self.metric_impl(op) {
    2312         2967 :             Some(x) => x,
    2313            8 :             None => return,
    2314              :         };
    2315         2967 :         self.metrics.call_end(&file_kind, &op_kind, track_bytes);
    2316         2975 :     }
    2317              : 
    2318              :     /// Close the upload queue for new operations and cancel queued operations.
    2319              :     ///
    2320              :     /// Use [`RemoteTimelineClient::shutdown`] for graceful stop.
    2321              :     ///
    2322              :     /// In-progress operations will still be running after this function returns.
    2323              :     /// Use `task_mgr::shutdown_tasks(Some(TaskKind::RemoteUploadTask), Some(self.tenant_shard_id), Some(timeline_id))`
    2324              :     /// to wait for them to complete, after calling this function.
    2325           18 :     pub(crate) fn stop(&self) {
    2326           18 :         // Whichever *task* for this RemoteTimelineClient grabs the mutex first will transition the queue
    2327           18 :         // into stopped state, thereby dropping all off the queued *ops* which haven't become *tasks* yet.
    2328           18 :         // The other *tasks* will come here and observe an already shut down queue and hence simply wrap up their business.
    2329           18 :         let mut guard = self.upload_queue.lock().unwrap();
    2330           18 :         self.stop_impl(&mut guard);
    2331           18 :     }
    2332              : 
    2333           18 :     fn stop_impl(&self, guard: &mut std::sync::MutexGuard<UploadQueue>) {
    2334           18 :         match &mut **guard {
    2335              :             UploadQueue::Uninitialized => {
    2336            0 :                 info!("UploadQueue is in state Uninitialized, nothing to do");
    2337            0 :                 **guard = UploadQueue::Stopped(UploadQueueStopped::Uninitialized);
    2338              :             }
    2339              :             UploadQueue::Stopped(_) => {
    2340              :                 // nothing to do
    2341            8 :                 info!("another concurrent task already shut down the queue");
    2342              :             }
    2343           10 :             UploadQueue::Initialized(initialized) => {
    2344           10 :                 info!("shutting down upload queue");
    2345              : 
    2346              :                 // Replace the queue with the Stopped state, taking ownership of the old
    2347              :                 // Initialized queue. We will do some checks on it, and then drop it.
    2348           10 :                 let qi = {
    2349              :                     // Here we preserve working version of the upload queue for possible use during deletions.
    2350              :                     // In-place replace of Initialized to Stopped can be done with the help of https://github.com/Sgeo/take_mut
    2351              :                     // but for this use case it doesnt really makes sense to bring unsafe code only for this usage point.
    2352              :                     // Deletion is not really perf sensitive so there shouldnt be any problems with cloning a fraction of it.
    2353           10 :                     let upload_queue_for_deletion = UploadQueueInitialized {
    2354           10 :                         task_counter: 0,
    2355           10 :                         dirty: initialized.dirty.clone(),
    2356           10 :                         clean: initialized.clean.clone(),
    2357           10 :                         latest_files_changes_since_metadata_upload_scheduled: 0,
    2358           10 :                         visible_remote_consistent_lsn: initialized
    2359           10 :                             .visible_remote_consistent_lsn
    2360           10 :                             .clone(),
    2361           10 :                         num_inprogress_layer_uploads: 0,
    2362           10 :                         num_inprogress_metadata_uploads: 0,
    2363           10 :                         num_inprogress_deletions: 0,
    2364           10 :                         inprogress_tasks: HashMap::default(),
    2365           10 :                         queued_operations: VecDeque::default(),
    2366           10 :                         #[cfg(feature = "testing")]
    2367           10 :                         dangling_files: HashMap::default(),
    2368           10 :                         blocked_deletions: Vec::new(),
    2369           10 :                         shutting_down: false,
    2370           10 :                         shutdown_ready: Arc::new(tokio::sync::Semaphore::new(0)),
    2371           10 :                         recently_deleted: HashSet::new(),
    2372           10 :                     };
    2373           10 : 
    2374           10 :                     let upload_queue = std::mem::replace(
    2375           10 :                         &mut **guard,
    2376           10 :                         UploadQueue::Stopped(UploadQueueStopped::Deletable(
    2377           10 :                             UploadQueueStoppedDeletable {
    2378           10 :                                 upload_queue_for_deletion,
    2379           10 :                                 deleted_at: SetDeletedFlagProgress::NotRunning,
    2380           10 :                             },
    2381           10 :                         )),
    2382           10 :                     );
    2383           10 :                     if let UploadQueue::Initialized(qi) = upload_queue {
    2384           10 :                         qi
    2385              :                     } else {
    2386            0 :                         unreachable!("we checked in the match above that it is Initialized");
    2387              :                     }
    2388              :                 };
    2389              : 
    2390              :                 // consistency check
    2391           10 :                 assert_eq!(
    2392           10 :                     qi.num_inprogress_layer_uploads
    2393           10 :                         + qi.num_inprogress_metadata_uploads
    2394           10 :                         + qi.num_inprogress_deletions,
    2395           10 :                     qi.inprogress_tasks.len()
    2396           10 :                 );
    2397              : 
    2398              :                 // We don't need to do anything here for in-progress tasks. They will finish
    2399              :                 // on their own, decrement the unfinished-task counter themselves, and observe
    2400              :                 // that the queue is Stopped.
    2401           10 :                 drop(qi.inprogress_tasks);
    2402              : 
    2403              :                 // Tear down queued ops
    2404           10 :                 for op in qi.queued_operations.into_iter() {
    2405            8 :                     self.metric_end(&op);
    2406            8 :                     // Dropping UploadOp::Barrier() here will make wait_completion() return with an Err()
    2407            8 :                     // which is exactly what we want to happen.
    2408            8 :                     drop(op);
    2409            8 :                 }
    2410              :             }
    2411              :         }
    2412           18 :     }
    2413              : 
    2414              :     /// Returns an accessor which will hold the UploadQueue mutex for accessing the upload queue
    2415              :     /// externally to RemoteTimelineClient.
    2416            0 :     pub(crate) fn initialized_upload_queue(
    2417            0 :         &self,
    2418            0 :     ) -> Result<UploadQueueAccessor<'_>, NotInitialized> {
    2419            0 :         let mut inner = self.upload_queue.lock().unwrap();
    2420            0 :         inner.initialized_mut()?;
    2421            0 :         Ok(UploadQueueAccessor { inner })
    2422            0 :     }
    2423              : 
    2424            8 :     pub(crate) fn no_pending_work(&self) -> bool {
    2425            8 :         let inner = self.upload_queue.lock().unwrap();
    2426            8 :         match &*inner {
    2427              :             UploadQueue::Uninitialized
    2428            0 :             | UploadQueue::Stopped(UploadQueueStopped::Uninitialized) => true,
    2429            8 :             UploadQueue::Stopped(UploadQueueStopped::Deletable(x)) => {
    2430            8 :                 x.upload_queue_for_deletion.no_pending_work()
    2431              :             }
    2432            0 :             UploadQueue::Initialized(x) => x.no_pending_work(),
    2433              :         }
    2434            8 :     }
    2435              : 
    2436              :     /// 'foreign' in the sense that it does not belong to this tenant shard.  This method
    2437              :     /// is used during GC for other shards to get the index of shard zero.
    2438            0 :     pub(crate) async fn download_foreign_index(
    2439            0 :         &self,
    2440            0 :         shard_number: ShardNumber,
    2441            0 :         cancel: &CancellationToken,
    2442            0 :     ) -> Result<(IndexPart, Generation, std::time::SystemTime), DownloadError> {
    2443            0 :         let foreign_shard_id = TenantShardId {
    2444            0 :             shard_number,
    2445            0 :             shard_count: self.tenant_shard_id.shard_count,
    2446            0 :             tenant_id: self.tenant_shard_id.tenant_id,
    2447            0 :         };
    2448            0 :         download_index_part(
    2449            0 :             &self.storage_impl,
    2450            0 :             &foreign_shard_id,
    2451            0 :             &self.timeline_id,
    2452            0 :             Generation::MAX,
    2453            0 :             cancel,
    2454            0 :         )
    2455            0 :         .await
    2456            0 :     }
    2457              : }
    2458              : 
    2459              : pub(crate) struct UploadQueueAccessor<'a> {
    2460              :     inner: std::sync::MutexGuard<'a, UploadQueue>,
    2461              : }
    2462              : 
    2463              : impl UploadQueueAccessor<'_> {
    2464            0 :     pub(crate) fn latest_uploaded_index_part(&self) -> &IndexPart {
    2465            0 :         match &*self.inner {
    2466            0 :             UploadQueue::Initialized(x) => &x.clean.0,
    2467              :             UploadQueue::Uninitialized | UploadQueue::Stopped(_) => {
    2468            0 :                 unreachable!("checked before constructing")
    2469              :             }
    2470              :         }
    2471            0 :     }
    2472              : }
    2473              : 
    2474            0 : pub fn remote_tenant_path(tenant_shard_id: &TenantShardId) -> RemotePath {
    2475            0 :     let path = format!("tenants/{tenant_shard_id}");
    2476            0 :     RemotePath::from_string(&path).expect("Failed to construct path")
    2477            0 : }
    2478              : 
    2479          590 : pub fn remote_tenant_manifest_path(
    2480          590 :     tenant_shard_id: &TenantShardId,
    2481          590 :     generation: Generation,
    2482          590 : ) -> RemotePath {
    2483          590 :     let path = format!(
    2484          590 :         "tenants/{tenant_shard_id}/tenant-manifest{}.json",
    2485          590 :         generation.get_suffix()
    2486          590 :     );
    2487          590 :     RemotePath::from_string(&path).expect("Failed to construct path")
    2488          590 : }
    2489              : 
    2490              : /// Prefix to all generations' manifest objects in a tenant shard
    2491          196 : pub fn remote_tenant_manifest_prefix(tenant_shard_id: &TenantShardId) -> RemotePath {
    2492          196 :     let path = format!("tenants/{tenant_shard_id}/tenant-manifest",);
    2493          196 :     RemotePath::from_string(&path).expect("Failed to construct path")
    2494          196 : }
    2495              : 
    2496          226 : pub fn remote_timelines_path(tenant_shard_id: &TenantShardId) -> RemotePath {
    2497          226 :     let path = format!("tenants/{tenant_shard_id}/{TIMELINES_SEGMENT_NAME}");
    2498          226 :     RemotePath::from_string(&path).expect("Failed to construct path")
    2499          226 : }
    2500              : 
    2501            0 : fn remote_timelines_path_unsharded(tenant_id: &TenantId) -> RemotePath {
    2502            0 :     let path = format!("tenants/{tenant_id}/{TIMELINES_SEGMENT_NAME}");
    2503            0 :     RemotePath::from_string(&path).expect("Failed to construct path")
    2504            0 : }
    2505              : 
    2506           30 : pub fn remote_timeline_path(
    2507           30 :     tenant_shard_id: &TenantShardId,
    2508           30 :     timeline_id: &TimelineId,
    2509           30 : ) -> RemotePath {
    2510           30 :     remote_timelines_path(tenant_shard_id).join(Utf8Path::new(&timeline_id.to_string()))
    2511           30 : }
    2512              : 
    2513              : /// Obtains the path of the given Layer in the remote
    2514              : ///
    2515              : /// Note that the shard component of a remote layer path is _not_ always the same
    2516              : /// as in the TenantShardId of the caller: tenants may reference layers from a different
    2517              : /// ShardIndex.  Use the ShardIndex from the layer's metadata.
    2518         1723 : pub fn remote_layer_path(
    2519         1723 :     tenant_id: &TenantId,
    2520         1723 :     timeline_id: &TimelineId,
    2521         1723 :     shard: ShardIndex,
    2522         1723 :     layer_file_name: &LayerName,
    2523         1723 :     generation: Generation,
    2524         1723 : ) -> RemotePath {
    2525         1723 :     // Generation-aware key format
    2526         1723 :     let path = format!(
    2527         1723 :         "tenants/{tenant_id}{0}/{TIMELINES_SEGMENT_NAME}/{timeline_id}/{1}{2}",
    2528         1723 :         shard.get_suffix(),
    2529         1723 :         layer_file_name,
    2530         1723 :         generation.get_suffix()
    2531         1723 :     );
    2532         1723 : 
    2533         1723 :     RemotePath::from_string(&path).expect("Failed to construct path")
    2534         1723 : }
    2535              : 
    2536              : /// Returns true if a and b have the same layer path within a tenant/timeline. This is essentially
    2537              : /// remote_layer_path(a) == remote_layer_path(b) without the string allocations.
    2538              : ///
    2539              : /// TODO: there should be a variant of LayerName for the physical path that contains information
    2540              : /// about the shard and generation, such that this could be replaced by a simple comparison.
    2541            4 : pub fn is_same_remote_layer_path(
    2542            4 :     aname: &LayerName,
    2543            4 :     ameta: &LayerFileMetadata,
    2544            4 :     bname: &LayerName,
    2545            4 :     bmeta: &LayerFileMetadata,
    2546            4 : ) -> bool {
    2547            4 :     // NB: don't assert remote_layer_path(a) == remote_layer_path(b); too expensive even for debug.
    2548            4 :     aname == bname && ameta.shard == bmeta.shard && ameta.generation == bmeta.generation
    2549            4 : }
    2550              : 
    2551            4 : pub fn remote_initdb_archive_path(tenant_id: &TenantId, timeline_id: &TimelineId) -> RemotePath {
    2552            4 :     RemotePath::from_string(&format!(
    2553            4 :         "tenants/{tenant_id}/{TIMELINES_SEGMENT_NAME}/{timeline_id}/{INITDB_PATH}"
    2554            4 :     ))
    2555            4 :     .expect("Failed to construct path")
    2556            4 : }
    2557              : 
    2558            2 : pub fn remote_initdb_preserved_archive_path(
    2559            2 :     tenant_id: &TenantId,
    2560            2 :     timeline_id: &TimelineId,
    2561            2 : ) -> RemotePath {
    2562            2 :     RemotePath::from_string(&format!(
    2563            2 :         "tenants/{tenant_id}/{TIMELINES_SEGMENT_NAME}/{timeline_id}/{INITDB_PRESERVED_PATH}"
    2564            2 :     ))
    2565            2 :     .expect("Failed to construct path")
    2566            2 : }
    2567              : 
    2568         1526 : pub fn remote_index_path(
    2569         1526 :     tenant_shard_id: &TenantShardId,
    2570         1526 :     timeline_id: &TimelineId,
    2571         1526 :     generation: Generation,
    2572         1526 : ) -> RemotePath {
    2573         1526 :     RemotePath::from_string(&format!(
    2574         1526 :         "tenants/{tenant_shard_id}/{TIMELINES_SEGMENT_NAME}/{timeline_id}/{0}{1}",
    2575         1526 :         IndexPart::FILE_NAME,
    2576         1526 :         generation.get_suffix()
    2577         1526 :     ))
    2578         1526 :     .expect("Failed to construct path")
    2579         1526 : }
    2580              : 
    2581            0 : pub(crate) fn remote_heatmap_path(tenant_shard_id: &TenantShardId) -> RemotePath {
    2582            0 :     RemotePath::from_string(&format!(
    2583            0 :         "tenants/{tenant_shard_id}/{TENANT_HEATMAP_BASENAME}"
    2584            0 :     ))
    2585            0 :     .expect("Failed to construct path")
    2586            0 : }
    2587              : 
    2588              : /// Given the key of an index, parse out the generation part of the name
    2589           18 : pub fn parse_remote_index_path(path: RemotePath) -> Option<Generation> {
    2590           18 :     let file_name = match path.get_path().file_name() {
    2591           18 :         Some(f) => f,
    2592              :         None => {
    2593              :             // Unexpected: we should be seeing index_part.json paths only
    2594            0 :             tracing::warn!("Malformed index key {}", path);
    2595            0 :             return None;
    2596              :         }
    2597              :     };
    2598              : 
    2599           18 :     match file_name.split_once('-') {
    2600           12 :         Some((_, gen_suffix)) => Generation::parse_suffix(gen_suffix),
    2601            6 :         None => None,
    2602              :     }
    2603           18 : }
    2604              : 
    2605              : /// Given the key of a tenant manifest, parse out the generation number
    2606            0 : pub fn parse_remote_tenant_manifest_path(path: RemotePath) -> Option<Generation> {
    2607              :     static RE: OnceLock<Regex> = OnceLock::new();
    2608            0 :     let re = RE.get_or_init(|| Regex::new(r".*tenant-manifest-([0-9a-f]{8}).json").unwrap());
    2609            0 :     re.captures(path.get_path().as_str())
    2610            0 :         .and_then(|c| c.get(1))
    2611            0 :         .and_then(|m| Generation::parse_suffix(m.as_str()))
    2612            0 : }
    2613              : 
    2614              : #[cfg(test)]
    2615              : mod tests {
    2616              :     use super::*;
    2617              :     use crate::{
    2618              :         context::RequestContext,
    2619              :         tenant::{
    2620              :             config::AttachmentMode,
    2621              :             harness::{TenantHarness, TIMELINE_ID},
    2622              :             storage_layer::layer::local_layer_path,
    2623              :             Tenant, Timeline,
    2624              :         },
    2625              :         DEFAULT_PG_VERSION,
    2626              :     };
    2627              : 
    2628              :     use std::collections::HashSet;
    2629              : 
    2630            8 :     pub(super) fn dummy_contents(name: &str) -> Vec<u8> {
    2631            8 :         format!("contents for {name}").into()
    2632            8 :     }
    2633              : 
    2634            2 :     pub(super) fn dummy_metadata(disk_consistent_lsn: Lsn) -> TimelineMetadata {
    2635            2 :         let metadata = TimelineMetadata::new(
    2636            2 :             disk_consistent_lsn,
    2637            2 :             None,
    2638            2 :             None,
    2639            2 :             Lsn(0),
    2640            2 :             Lsn(0),
    2641            2 :             Lsn(0),
    2642            2 :             // Any version will do
    2643            2 :             // but it should be consistent with the one in the tests
    2644            2 :             crate::DEFAULT_PG_VERSION,
    2645            2 :         );
    2646            2 : 
    2647            2 :         // go through serialize + deserialize to fix the header, including checksum
    2648            2 :         TimelineMetadata::from_bytes(&metadata.to_bytes().unwrap()).unwrap()
    2649            2 :     }
    2650              : 
    2651            2 :     fn assert_file_list(a: &HashSet<LayerName>, b: &[&str]) {
    2652            6 :         let mut avec: Vec<String> = a.iter().map(|x| x.to_string()).collect();
    2653            2 :         avec.sort();
    2654            2 : 
    2655            2 :         let mut bvec = b.to_vec();
    2656            2 :         bvec.sort_unstable();
    2657            2 : 
    2658            2 :         assert_eq!(avec, bvec);
    2659            2 :     }
    2660              : 
    2661            4 :     fn assert_remote_files(expected: &[&str], remote_path: &Utf8Path, generation: Generation) {
    2662            4 :         let mut expected: Vec<String> = expected
    2663            4 :             .iter()
    2664           16 :             .map(|x| format!("{}{}", x, generation.get_suffix()))
    2665            4 :             .collect();
    2666            4 :         expected.sort();
    2667            4 : 
    2668            4 :         let mut found: Vec<String> = Vec::new();
    2669           16 :         for entry in std::fs::read_dir(remote_path).unwrap().flatten() {
    2670           16 :             let entry_name = entry.file_name();
    2671           16 :             let fname = entry_name.to_str().unwrap();
    2672           16 :             found.push(String::from(fname));
    2673           16 :         }
    2674            4 :         found.sort();
    2675            4 : 
    2676            4 :         assert_eq!(found, expected);
    2677            4 :     }
    2678              : 
    2679              :     struct TestSetup {
    2680              :         harness: TenantHarness,
    2681              :         tenant: Arc<Tenant>,
    2682              :         timeline: Arc<Timeline>,
    2683              :         tenant_ctx: RequestContext,
    2684              :     }
    2685              : 
    2686              :     impl TestSetup {
    2687            8 :         async fn new(test_name: &str) -> anyhow::Result<Self> {
    2688            8 :             let test_name = Box::leak(Box::new(format!("remote_timeline_client__{test_name}")));
    2689            8 :             let harness = TenantHarness::create(test_name).await?;
    2690            8 :             let (tenant, ctx) = harness.load().await;
    2691              : 
    2692            8 :             let timeline = tenant
    2693            8 :                 .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)
    2694            8 :                 .await?;
    2695              : 
    2696            8 :             Ok(Self {
    2697            8 :                 harness,
    2698            8 :                 tenant,
    2699            8 :                 timeline,
    2700            8 :                 tenant_ctx: ctx,
    2701            8 :             })
    2702            8 :         }
    2703              : 
    2704              :         /// Construct a RemoteTimelineClient in an arbitrary generation
    2705           10 :         fn build_client(&self, generation: Generation) -> Arc<RemoteTimelineClient> {
    2706           10 :             let location_conf = AttachedLocationConfig {
    2707           10 :                 generation,
    2708           10 :                 attach_mode: AttachmentMode::Single,
    2709           10 :             };
    2710           10 :             Arc::new(RemoteTimelineClient {
    2711           10 :                 conf: self.harness.conf,
    2712           10 :                 runtime: tokio::runtime::Handle::current(),
    2713           10 :                 tenant_shard_id: self.harness.tenant_shard_id,
    2714           10 :                 timeline_id: TIMELINE_ID,
    2715           10 :                 generation,
    2716           10 :                 storage_impl: self.harness.remote_storage.clone(),
    2717           10 :                 deletion_queue_client: self.harness.deletion_queue.new_client(),
    2718           10 :                 upload_queue: Mutex::new(UploadQueue::Uninitialized),
    2719           10 :                 metrics: Arc::new(RemoteTimelineClientMetrics::new(
    2720           10 :                     &self.harness.tenant_shard_id,
    2721           10 :                     &TIMELINE_ID,
    2722           10 :                 )),
    2723           10 :                 config: std::sync::RwLock::new(RemoteTimelineClientConfig::from(&location_conf)),
    2724           10 :                 cancel: CancellationToken::new(),
    2725           10 :             })
    2726           10 :         }
    2727              : 
    2728              :         /// A tracing::Span that satisfies remote_timeline_client methods that assert tenant_id
    2729              :         /// and timeline_id are present.
    2730            6 :         fn span(&self) -> tracing::Span {
    2731            6 :             tracing::info_span!(
    2732              :                 "test",
    2733              :                 tenant_id = %self.harness.tenant_shard_id.tenant_id,
    2734            0 :                 shard_id = %self.harness.tenant_shard_id.shard_slug(),
    2735              :                 timeline_id = %TIMELINE_ID
    2736              :             )
    2737            6 :         }
    2738              :     }
    2739              : 
    2740              :     // Test scheduling
    2741              :     #[tokio::test]
    2742            2 :     async fn upload_scheduling() {
    2743            2 :         // Test outline:
    2744            2 :         //
    2745            2 :         // Schedule upload of a bunch of layers. Check that they are started immediately, not queued
    2746            2 :         // Schedule upload of index. Check that it is queued
    2747            2 :         // let the layer file uploads finish. Check that the index-upload is now started
    2748            2 :         // let the index-upload finish.
    2749            2 :         //
    2750            2 :         // Download back the index.json. Check that the list of files is correct
    2751            2 :         //
    2752            2 :         // Schedule upload. Schedule deletion. Check that the deletion is queued
    2753            2 :         // let upload finish. Check that deletion is now started
    2754            2 :         // Schedule another deletion. Check that it's launched immediately.
    2755            2 :         // Schedule index upload. Check that it's queued
    2756            2 : 
    2757            2 :         let test_setup = TestSetup::new("upload_scheduling").await.unwrap();
    2758            2 :         let span = test_setup.span();
    2759            2 :         let _guard = span.enter();
    2760            2 : 
    2761            2 :         let TestSetup {
    2762            2 :             harness,
    2763            2 :             tenant: _tenant,
    2764            2 :             timeline,
    2765            2 :             tenant_ctx: _tenant_ctx,
    2766            2 :         } = test_setup;
    2767            2 : 
    2768            2 :         let client = &timeline.remote_client;
    2769            2 : 
    2770            2 :         // Download back the index.json, and check that the list of files is correct
    2771            2 :         let initial_index_part = match client
    2772            2 :             .download_index_file(&CancellationToken::new())
    2773            2 :             .await
    2774            2 :             .unwrap()
    2775            2 :         {
    2776            2 :             MaybeDeletedIndexPart::IndexPart(index_part) => index_part,
    2777            2 :             MaybeDeletedIndexPart::Deleted(_) => panic!("unexpectedly got deleted index part"),
    2778            2 :         };
    2779            2 :         let initial_layers = initial_index_part
    2780            2 :             .layer_metadata
    2781            2 :             .keys()
    2782            2 :             .map(|f| f.to_owned())
    2783            2 :             .collect::<HashSet<LayerName>>();
    2784            2 :         let initial_layer = {
    2785            2 :             assert!(initial_layers.len() == 1);
    2786            2 :             initial_layers.into_iter().next().unwrap()
    2787            2 :         };
    2788            2 : 
    2789            2 :         let timeline_path = harness.timeline_path(&TIMELINE_ID);
    2790            2 : 
    2791            2 :         println!("workdir: {}", harness.conf.workdir);
    2792            2 : 
    2793            2 :         let remote_timeline_dir = harness
    2794            2 :             .remote_fs_dir
    2795            2 :             .join(timeline_path.strip_prefix(&harness.conf.workdir).unwrap());
    2796            2 :         println!("remote_timeline_dir: {remote_timeline_dir}");
    2797            2 : 
    2798            2 :         let generation = harness.generation;
    2799            2 :         let shard = harness.shard;
    2800            2 : 
    2801            2 :         // Create a couple of dummy files,  schedule upload for them
    2802            2 : 
    2803            2 :         let layers = [
    2804            2 :             ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51".parse().unwrap(), dummy_contents("foo")),
    2805            2 :             ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D9-00000000016B5A52".parse().unwrap(), dummy_contents("bar")),
    2806            2 :             ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59DA-00000000016B5A53".parse().unwrap(), dummy_contents("baz"))
    2807            2 :         ]
    2808            2 :         .into_iter()
    2809            6 :         .map(|(name, contents): (LayerName, Vec<u8>)| {
    2810            6 : 
    2811            6 :             let local_path = local_layer_path(
    2812            6 :                 harness.conf,
    2813            6 :                 &timeline.tenant_shard_id,
    2814            6 :                 &timeline.timeline_id,
    2815            6 :                 &name,
    2816            6 :                 &generation,
    2817            6 :             );
    2818            6 :             std::fs::write(&local_path, &contents).unwrap();
    2819            6 : 
    2820            6 :             Layer::for_resident(
    2821            6 :                 harness.conf,
    2822            6 :                 &timeline,
    2823            6 :                 local_path,
    2824            6 :                 name,
    2825            6 :                 LayerFileMetadata::new(contents.len() as u64, generation, shard),
    2826            6 :             )
    2827            6 :         }).collect::<Vec<_>>();
    2828            2 : 
    2829            2 :         client
    2830            2 :             .schedule_layer_file_upload(layers[0].clone())
    2831            2 :             .unwrap();
    2832            2 :         client
    2833            2 :             .schedule_layer_file_upload(layers[1].clone())
    2834            2 :             .unwrap();
    2835            2 : 
    2836            2 :         // Check that they are started immediately, not queued
    2837            2 :         //
    2838            2 :         // this works because we running within block_on, so any futures are now queued up until
    2839            2 :         // our next await point.
    2840            2 :         {
    2841            2 :             let mut guard = client.upload_queue.lock().unwrap();
    2842            2 :             let upload_queue = guard.initialized_mut().unwrap();
    2843            2 :             assert!(upload_queue.queued_operations.is_empty());
    2844            2 :             assert!(upload_queue.inprogress_tasks.len() == 2);
    2845            2 :             assert!(upload_queue.num_inprogress_layer_uploads == 2);
    2846            2 : 
    2847            2 :             // also check that `latest_file_changes` was updated
    2848            2 :             assert!(upload_queue.latest_files_changes_since_metadata_upload_scheduled == 2);
    2849            2 :         }
    2850            2 : 
    2851            2 :         // Schedule upload of index. Check that it is queued
    2852            2 :         let metadata = dummy_metadata(Lsn(0x20));
    2853            2 :         client
    2854            2 :             .schedule_index_upload_for_full_metadata_update(&metadata)
    2855            2 :             .unwrap();
    2856            2 :         {
    2857            2 :             let mut guard = client.upload_queue.lock().unwrap();
    2858            2 :             let upload_queue = guard.initialized_mut().unwrap();
    2859            2 :             assert!(upload_queue.queued_operations.len() == 1);
    2860            2 :             assert!(upload_queue.latest_files_changes_since_metadata_upload_scheduled == 0);
    2861            2 :         }
    2862            2 : 
    2863            2 :         // Wait for the uploads to finish
    2864            2 :         client.wait_completion().await.unwrap();
    2865            2 :         {
    2866            2 :             let mut guard = client.upload_queue.lock().unwrap();
    2867            2 :             let upload_queue = guard.initialized_mut().unwrap();
    2868            2 : 
    2869            2 :             assert!(upload_queue.queued_operations.is_empty());
    2870            2 :             assert!(upload_queue.inprogress_tasks.is_empty());
    2871            2 :         }
    2872            2 : 
    2873            2 :         // Download back the index.json, and check that the list of files is correct
    2874            2 :         let index_part = match client
    2875            2 :             .download_index_file(&CancellationToken::new())
    2876            2 :             .await
    2877            2 :             .unwrap()
    2878            2 :         {
    2879            2 :             MaybeDeletedIndexPart::IndexPart(index_part) => index_part,
    2880            2 :             MaybeDeletedIndexPart::Deleted(_) => panic!("unexpectedly got deleted index part"),
    2881            2 :         };
    2882            2 : 
    2883            2 :         assert_file_list(
    2884            2 :             &index_part
    2885            2 :                 .layer_metadata
    2886            2 :                 .keys()
    2887            6 :                 .map(|f| f.to_owned())
    2888            2 :                 .collect(),
    2889            2 :             &[
    2890            2 :                 &initial_layer.to_string(),
    2891            2 :                 &layers[0].layer_desc().layer_name().to_string(),
    2892            2 :                 &layers[1].layer_desc().layer_name().to_string(),
    2893            2 :             ],
    2894            2 :         );
    2895            2 :         assert_eq!(index_part.metadata, metadata);
    2896            2 : 
    2897            2 :         // Schedule upload and then a deletion. Check that the deletion is queued
    2898            2 :         client
    2899            2 :             .schedule_layer_file_upload(layers[2].clone())
    2900            2 :             .unwrap();
    2901            2 : 
    2902            2 :         // this is no longer consistent with how deletion works with Layer::drop, but in this test
    2903            2 :         // keep using schedule_layer_file_deletion because we don't have a way to wait for the
    2904            2 :         // spawn_blocking started by the drop.
    2905            2 :         client
    2906            2 :             .schedule_layer_file_deletion(&[layers[0].layer_desc().layer_name()])
    2907            2 :             .unwrap();
    2908            2 :         {
    2909            2 :             let mut guard = client.upload_queue.lock().unwrap();
    2910            2 :             let upload_queue = guard.initialized_mut().unwrap();
    2911            2 : 
    2912            2 :             // Deletion schedules upload of the index file, and the file deletion itself
    2913            2 :             assert_eq!(upload_queue.queued_operations.len(), 2);
    2914            2 :             assert_eq!(upload_queue.inprogress_tasks.len(), 1);
    2915            2 :             assert_eq!(upload_queue.num_inprogress_layer_uploads, 1);
    2916            2 :             assert_eq!(upload_queue.num_inprogress_deletions, 0);
    2917            2 :             assert_eq!(
    2918            2 :                 upload_queue.latest_files_changes_since_metadata_upload_scheduled,
    2919            2 :                 0
    2920            2 :             );
    2921            2 :         }
    2922            2 :         assert_remote_files(
    2923            2 :             &[
    2924            2 :                 &initial_layer.to_string(),
    2925            2 :                 &layers[0].layer_desc().layer_name().to_string(),
    2926            2 :                 &layers[1].layer_desc().layer_name().to_string(),
    2927            2 :                 "index_part.json",
    2928            2 :             ],
    2929            2 :             &remote_timeline_dir,
    2930            2 :             generation,
    2931            2 :         );
    2932            2 : 
    2933            2 :         // Finish them
    2934            2 :         client.wait_completion().await.unwrap();
    2935            2 :         harness.deletion_queue.pump().await;
    2936            2 : 
    2937            2 :         assert_remote_files(
    2938            2 :             &[
    2939            2 :                 &initial_layer.to_string(),
    2940            2 :                 &layers[1].layer_desc().layer_name().to_string(),
    2941            2 :                 &layers[2].layer_desc().layer_name().to_string(),
    2942            2 :                 "index_part.json",
    2943            2 :             ],
    2944            2 :             &remote_timeline_dir,
    2945            2 :             generation,
    2946            2 :         );
    2947            2 :     }
    2948              : 
    2949              :     #[tokio::test]
    2950            2 :     async fn bytes_unfinished_gauge_for_layer_file_uploads() {
    2951            2 :         // Setup
    2952            2 : 
    2953            2 :         let TestSetup {
    2954            2 :             harness,
    2955            2 :             tenant: _tenant,
    2956            2 :             timeline,
    2957            2 :             ..
    2958            2 :         } = TestSetup::new("metrics").await.unwrap();
    2959            2 :         let client = &timeline.remote_client;
    2960            2 : 
    2961            2 :         let layer_file_name_1: LayerName = "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51".parse().unwrap();
    2962            2 :         let local_path = local_layer_path(
    2963            2 :             harness.conf,
    2964            2 :             &timeline.tenant_shard_id,
    2965            2 :             &timeline.timeline_id,
    2966            2 :             &layer_file_name_1,
    2967            2 :             &harness.generation,
    2968            2 :         );
    2969            2 :         let content_1 = dummy_contents("foo");
    2970            2 :         std::fs::write(&local_path, &content_1).unwrap();
    2971            2 : 
    2972            2 :         let layer_file_1 = Layer::for_resident(
    2973            2 :             harness.conf,
    2974            2 :             &timeline,
    2975            2 :             local_path,
    2976            2 :             layer_file_name_1.clone(),
    2977            2 :             LayerFileMetadata::new(content_1.len() as u64, harness.generation, harness.shard),
    2978            2 :         );
    2979            2 : 
    2980            2 :         #[derive(Debug, PartialEq, Clone, Copy)]
    2981            2 :         struct BytesStartedFinished {
    2982            2 :             started: Option<usize>,
    2983            2 :             finished: Option<usize>,
    2984            2 :         }
    2985            2 :         impl std::ops::Add for BytesStartedFinished {
    2986            2 :             type Output = Self;
    2987            4 :             fn add(self, rhs: Self) -> Self::Output {
    2988            4 :                 Self {
    2989            4 :                     started: self.started.map(|v| v + rhs.started.unwrap_or(0)),
    2990            4 :                     finished: self.finished.map(|v| v + rhs.finished.unwrap_or(0)),
    2991            4 :                 }
    2992            4 :             }
    2993            2 :         }
    2994            6 :         let get_bytes_started_stopped = || {
    2995            6 :             let started = client
    2996            6 :                 .metrics
    2997            6 :                 .get_bytes_started_counter_value(&RemoteOpFileKind::Layer, &RemoteOpKind::Upload)
    2998            6 :                 .map(|v| v.try_into().unwrap());
    2999            6 :             let stopped = client
    3000            6 :                 .metrics
    3001            6 :                 .get_bytes_finished_counter_value(&RemoteOpFileKind::Layer, &RemoteOpKind::Upload)
    3002            6 :                 .map(|v| v.try_into().unwrap());
    3003            6 :             BytesStartedFinished {
    3004            6 :                 started,
    3005            6 :                 finished: stopped,
    3006            6 :             }
    3007            6 :         };
    3008            2 : 
    3009            2 :         // Test
    3010            2 :         tracing::info!("now doing actual test");
    3011            2 : 
    3012            2 :         let actual_a = get_bytes_started_stopped();
    3013            2 : 
    3014            2 :         client
    3015            2 :             .schedule_layer_file_upload(layer_file_1.clone())
    3016            2 :             .unwrap();
    3017            2 : 
    3018            2 :         let actual_b = get_bytes_started_stopped();
    3019            2 : 
    3020            2 :         client.wait_completion().await.unwrap();
    3021            2 : 
    3022            2 :         let actual_c = get_bytes_started_stopped();
    3023            2 : 
    3024            2 :         // Validate
    3025            2 : 
    3026            2 :         let expected_b = actual_a
    3027            2 :             + BytesStartedFinished {
    3028            2 :                 started: Some(content_1.len()),
    3029            2 :                 // assert that the _finished metric is created eagerly so that subtractions work on first sample
    3030            2 :                 finished: Some(0),
    3031            2 :             };
    3032            2 :         assert_eq!(actual_b, expected_b);
    3033            2 : 
    3034            2 :         let expected_c = actual_a
    3035            2 :             + BytesStartedFinished {
    3036            2 :                 started: Some(content_1.len()),
    3037            2 :                 finished: Some(content_1.len()),
    3038            2 :             };
    3039            2 :         assert_eq!(actual_c, expected_c);
    3040            2 :     }
    3041              : 
    3042           12 :     async fn inject_index_part(test_state: &TestSetup, generation: Generation) -> IndexPart {
    3043           12 :         // An empty IndexPart, just sufficient to ensure deserialization will succeed
    3044           12 :         let example_index_part = IndexPart::example();
    3045           12 : 
    3046           12 :         let index_part_bytes = serde_json::to_vec(&example_index_part).unwrap();
    3047           12 : 
    3048           12 :         let index_path = test_state.harness.remote_fs_dir.join(
    3049           12 :             remote_index_path(
    3050           12 :                 &test_state.harness.tenant_shard_id,
    3051           12 :                 &TIMELINE_ID,
    3052           12 :                 generation,
    3053           12 :             )
    3054           12 :             .get_path(),
    3055           12 :         );
    3056           12 : 
    3057           12 :         std::fs::create_dir_all(index_path.parent().unwrap())
    3058           12 :             .expect("creating test dir should work");
    3059           12 : 
    3060           12 :         eprintln!("Writing {index_path}");
    3061           12 :         std::fs::write(&index_path, index_part_bytes).unwrap();
    3062           12 :         example_index_part
    3063           12 :     }
    3064              : 
    3065              :     /// Assert that when a RemoteTimelineclient in generation `get_generation` fetches its
    3066              :     /// index, the IndexPart returned is equal to `expected`
    3067           10 :     async fn assert_got_index_part(
    3068           10 :         test_state: &TestSetup,
    3069           10 :         get_generation: Generation,
    3070           10 :         expected: &IndexPart,
    3071           10 :     ) {
    3072           10 :         let client = test_state.build_client(get_generation);
    3073              : 
    3074           10 :         let download_r = client
    3075           10 :             .download_index_file(&CancellationToken::new())
    3076           10 :             .await
    3077           10 :             .expect("download should always succeed");
    3078           10 :         assert!(matches!(download_r, MaybeDeletedIndexPart::IndexPart(_)));
    3079           10 :         match download_r {
    3080           10 :             MaybeDeletedIndexPart::IndexPart(index_part) => {
    3081           10 :                 assert_eq!(&index_part, expected);
    3082              :             }
    3083            0 :             MaybeDeletedIndexPart::Deleted(_index_part) => panic!("Test doesn't set deleted_at"),
    3084              :         }
    3085           10 :     }
    3086              : 
    3087              :     #[tokio::test]
    3088            2 :     async fn index_part_download_simple() -> anyhow::Result<()> {
    3089            2 :         let test_state = TestSetup::new("index_part_download_simple").await.unwrap();
    3090            2 :         let span = test_state.span();
    3091            2 :         let _guard = span.enter();
    3092            2 : 
    3093            2 :         // Simple case: we are in generation N, load the index from generation N - 1
    3094            2 :         let generation_n = 5;
    3095            2 :         let injected = inject_index_part(&test_state, Generation::new(generation_n - 1)).await;
    3096            2 : 
    3097            2 :         assert_got_index_part(&test_state, Generation::new(generation_n), &injected).await;
    3098            2 : 
    3099            2 :         Ok(())
    3100            2 :     }
    3101              : 
    3102              :     #[tokio::test]
    3103            2 :     async fn index_part_download_ordering() -> anyhow::Result<()> {
    3104            2 :         let test_state = TestSetup::new("index_part_download_ordering")
    3105            2 :             .await
    3106            2 :             .unwrap();
    3107            2 : 
    3108            2 :         let span = test_state.span();
    3109            2 :         let _guard = span.enter();
    3110            2 : 
    3111            2 :         // A generation-less IndexPart exists in the bucket, we should find it
    3112            2 :         let generation_n = 5;
    3113            2 :         let injected_none = inject_index_part(&test_state, Generation::none()).await;
    3114            2 :         assert_got_index_part(&test_state, Generation::new(generation_n), &injected_none).await;
    3115            2 : 
    3116            2 :         // If a more recent-than-none generation exists, we should prefer to load that
    3117            2 :         let injected_1 = inject_index_part(&test_state, Generation::new(1)).await;
    3118            2 :         assert_got_index_part(&test_state, Generation::new(generation_n), &injected_1).await;
    3119            2 : 
    3120            2 :         // If a more-recent-than-me generation exists, we should ignore it.
    3121            2 :         let _injected_10 = inject_index_part(&test_state, Generation::new(10)).await;
    3122            2 :         assert_got_index_part(&test_state, Generation::new(generation_n), &injected_1).await;
    3123            2 : 
    3124            2 :         // If a directly previous generation exists, _and_ an index exists in my own
    3125            2 :         // generation, I should prefer my own generation.
    3126            2 :         let _injected_prev =
    3127            2 :             inject_index_part(&test_state, Generation::new(generation_n - 1)).await;
    3128            2 :         let injected_current = inject_index_part(&test_state, Generation::new(generation_n)).await;
    3129            2 :         assert_got_index_part(
    3130            2 :             &test_state,
    3131            2 :             Generation::new(generation_n),
    3132            2 :             &injected_current,
    3133            2 :         )
    3134            2 :         .await;
    3135            2 : 
    3136            2 :         Ok(())
    3137            2 :     }
    3138              : }
        

Generated by: LCOV version 2.1-beta