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

Generated by: LCOV version 2.1-beta