LCOV - code coverage report
Current view: top level - pageserver/src/tenant - remote_timeline_client.rs (source / functions) Coverage Total Hit
Test: 1b0a6a0c05cee5a7de360813c8034804e105ce1c.info Lines: 63.9 % 2084 1332
Test Date: 2025-03-12 00:01:28 Functions: 51.5 % 196 101

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

Generated by: LCOV version 2.1-beta