LCOV - code coverage report
Current view: top level - pageserver/src/tenant - upload_queue.rs (source / functions) Coverage Total Hit
Test: 5fe7fa8d483b39476409aee736d6d5e32728bfac.info Lines: 94.6 % 959 907
Test Date: 2025-03-12 16:10:49 Functions: 92.6 % 81 75

            Line data    Source code
       1              : use std::collections::{HashMap, HashSet, VecDeque};
       2              : use std::fmt::Debug;
       3              : use std::sync::Arc;
       4              : use std::sync::atomic::AtomicU32;
       5              : 
       6              : use chrono::NaiveDateTime;
       7              : use once_cell::sync::Lazy;
       8              : use tracing::info;
       9              : use utils::generation::Generation;
      10              : use utils::lsn::{AtomicLsn, Lsn};
      11              : 
      12              : use super::remote_timeline_client::is_same_remote_layer_path;
      13              : use super::storage_layer::{AsLayerDesc as _, LayerName, ResidentLayer};
      14              : use crate::tenant::metadata::TimelineMetadata;
      15              : use crate::tenant::remote_timeline_client::index::{IndexPart, LayerFileMetadata};
      16              : 
      17              : /// Kill switch for upload queue reordering in case it causes problems.
      18              : /// TODO: remove this once we have confidence in it.
      19              : static DISABLE_UPLOAD_QUEUE_REORDERING: Lazy<bool> =
      20          397 :     Lazy::new(|| std::env::var("DISABLE_UPLOAD_QUEUE_REORDERING").as_deref() == Ok("true"));
      21              : 
      22              : /// Kill switch for index upload coalescing in case it causes problems.
      23              : /// TODO: remove this once we have confidence in it.
      24              : static DISABLE_UPLOAD_QUEUE_INDEX_COALESCING: Lazy<bool> =
      25           15 :     Lazy::new(|| std::env::var("DISABLE_UPLOAD_QUEUE_INDEX_COALESCING").as_deref() == Ok("true"));
      26              : 
      27              : // clippy warns that Uninitialized is much smaller than Initialized, which wastes
      28              : // memory for Uninitialized variants. Doesn't matter in practice, there are not
      29              : // that many upload queues in a running pageserver, and most of them are initialized
      30              : // anyway.
      31              : #[allow(clippy::large_enum_variant)]
      32              : pub enum UploadQueue {
      33              :     Uninitialized,
      34              :     Initialized(UploadQueueInitialized),
      35              :     Stopped(UploadQueueStopped),
      36              : }
      37              : 
      38              : impl UploadQueue {
      39            0 :     pub fn as_str(&self) -> &'static str {
      40            0 :         match self {
      41            0 :             UploadQueue::Uninitialized => "Uninitialized",
      42            0 :             UploadQueue::Initialized(_) => "Initialized",
      43            0 :             UploadQueue::Stopped(_) => "Stopped",
      44              :         }
      45            0 :     }
      46              : }
      47              : 
      48              : #[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
      49              : pub enum OpType {
      50              :     MayReorder,
      51              :     FlushDeletion,
      52              : }
      53              : 
      54              : /// This keeps track of queued and in-progress tasks.
      55              : pub struct UploadQueueInitialized {
      56              :     /// Maximum number of inprogress tasks to schedule. 0 is no limit.
      57              :     pub(crate) inprogress_limit: usize,
      58              : 
      59              :     /// Counter to assign task IDs
      60              :     pub(crate) task_counter: u64,
      61              : 
      62              :     /// The next uploaded index_part.json; assumed to be dirty.
      63              :     ///
      64              :     /// Should not be read, directly except for layer file updates. Instead you should add a
      65              :     /// projected field.
      66              :     pub(crate) dirty: IndexPart,
      67              : 
      68              :     /// The latest remote persisted IndexPart.
      69              :     ///
      70              :     /// Each completed metadata upload will update this. The second item is the task_id which last
      71              :     /// updated the value, used to ensure we never store an older value over a newer one.
      72              :     pub(crate) clean: (IndexPart, Option<u64>),
      73              : 
      74              :     /// How many file uploads or deletions been scheduled, since the
      75              :     /// last (scheduling of) metadata index upload?
      76              :     pub(crate) latest_files_changes_since_metadata_upload_scheduled: u64,
      77              : 
      78              :     /// The Lsn is only updated after our generation has been validated with
      79              :     /// the control plane (unlesss a timeline's generation is None, in which case
      80              :     /// we skip validation)
      81              :     pub(crate) visible_remote_consistent_lsn: Arc<AtomicLsn>,
      82              : 
      83              :     /// Tasks that are currently in-progress. In-progress means that a tokio Task
      84              :     /// has been launched for it. An in-progress task can be busy uploading, but it can
      85              :     /// also be waiting on the `concurrency_limiter` Semaphore in S3Bucket, or it can
      86              :     /// be waiting for retry in `exponential_backoff`.
      87              :     pub inprogress_tasks: HashMap<u64, Arc<UploadTask>>,
      88              : 
      89              :     /// Queued operations that have not been launched yet. They might depend on previous
      90              :     /// tasks to finish. For example, metadata upload cannot be performed before all
      91              :     /// preceding layer file uploads have completed.
      92              :     pub queued_operations: VecDeque<UploadOp>,
      93              : 
      94              :     /// Files which have been unlinked but not yet had scheduled a deletion for. Only kept around
      95              :     /// for error logging.
      96              :     ///
      97              :     /// Putting this behind a testing feature to catch problems in tests, but assuming we could have a
      98              :     /// bug causing leaks, then it's better to not leave this enabled for production builds.
      99              :     #[cfg(feature = "testing")]
     100              :     pub(crate) dangling_files: HashMap<LayerName, Generation>,
     101              : 
     102              :     /// Ensure we order file operations correctly.
     103              :     pub(crate) recently_deleted: HashSet<(LayerName, Generation)>,
     104              : 
     105              :     /// Deletions that are blocked by the tenant configuration
     106              :     pub(crate) blocked_deletions: Vec<Delete>,
     107              : 
     108              :     /// Set to true when we have inserted the `UploadOp::Shutdown` into the `inprogress_tasks`.
     109              :     pub(crate) shutting_down: bool,
     110              : 
     111              :     /// Permitless semaphore on which any number of `RemoteTimelineClient::shutdown` futures can
     112              :     /// wait on until one of them stops the queue. The semaphore is closed when
     113              :     /// `RemoteTimelineClient::launch_queued_tasks` encounters `UploadOp::Shutdown`.
     114              :     pub(crate) shutdown_ready: Arc<tokio::sync::Semaphore>,
     115              : }
     116              : 
     117              : impl UploadQueueInitialized {
     118           16 :     pub(super) fn no_pending_work(&self) -> bool {
     119           16 :         self.inprogress_tasks.is_empty() && self.queued_operations.is_empty()
     120           16 :     }
     121              : 
     122            0 :     pub(super) fn get_last_remote_consistent_lsn_visible(&self) -> Lsn {
     123            0 :         self.visible_remote_consistent_lsn.load()
     124            0 :     }
     125              : 
     126            0 :     pub(super) fn get_last_remote_consistent_lsn_projected(&self) -> Option<Lsn> {
     127            0 :         let lsn = self.clean.0.metadata.disk_consistent_lsn();
     128            0 :         self.clean.1.map(|_| lsn)
     129            0 :     }
     130              : 
     131              :     /// Returns and removes the next ready operation from the queue, if any. This isn't necessarily
     132              :     /// the first operation in the queue, to avoid head-of-line blocking -- an operation can jump
     133              :     /// the queue if it doesn't conflict with operations ahead of it.
     134              :     ///
     135              :     /// Also returns any operations that were coalesced into this one, e.g. multiple index uploads.
     136              :     ///
     137              :     /// None may be returned even if the queue isn't empty, if no operations are ready yet.
     138              :     ///
     139              :     /// NB: this is quadratic, but queues are expected to be small, and bounded by inprogress_limit.
     140        26818 :     pub fn next_ready(&mut self) -> Option<(UploadOp, Vec<UploadOp>)> {
     141        26818 :         // If inprogress_tasks is already at limit, don't schedule anything more.
     142        26818 :         if self.inprogress_limit > 0 && self.inprogress_tasks.len() >= self.inprogress_limit {
     143           12 :             return None;
     144        26806 :         }
     145              : 
     146        51881 :         for (i, candidate) in self.queued_operations.iter().enumerate() {
     147              :             // If this candidate is ready, go for it. Otherwise, try the next one.
     148        51881 :             if self.is_ready(i) {
     149              :                 // Shutdown operations are left at the head of the queue, to prevent further
     150              :                 // operations from starting. Signal that we're ready to shut down.
     151        10146 :                 if matches!(candidate, UploadOp::Shutdown) {
     152           20 :                     assert!(self.inprogress_tasks.is_empty(), "shutdown with tasks");
     153           20 :                     assert_eq!(i, 0, "shutdown not at head of queue");
     154           20 :                     self.shutdown_ready.close();
     155           20 :                     return None;
     156        10126 :                 }
     157        10126 : 
     158        10126 :                 let mut op = self.queued_operations.remove(i).expect("i can't disappear");
     159        10126 : 
     160        10126 :                 // Coalesce any back-to-back index uploads by only uploading the newest one that's
     161        10126 :                 // ready. This typically happens with layer/index/layer/index/... sequences, where
     162        10126 :                 // the layers bypass the indexes, leaving the indexes queued.
     163        10126 :                 //
     164        10126 :                 // If other operations are interleaved between index uploads we don't try to
     165        10126 :                 // coalesce them, since we may as well update the index concurrently with them.
     166        10126 :                 // This keeps the index fresh and avoids starvation.
     167        10126 :                 //
     168        10126 :                 // NB: we assume that all uploaded indexes have the same remote path. This
     169        10126 :                 // is true at the time of writing: the path only depends on the tenant,
     170        10126 :                 // timeline and generation, all of which are static for a timeline instance.
     171        10126 :                 // Otherwise, we must be careful not to coalesce different paths.
     172        10126 :                 let mut coalesced_ops = Vec::new();
     173        10126 :                 if matches!(op, UploadOp::UploadMetadata { .. }) {
     174         3064 :                     while let Some(UploadOp::UploadMetadata { .. }) = self.queued_operations.get(i)
     175              :                     {
     176           29 :                         if *DISABLE_UPLOAD_QUEUE_INDEX_COALESCING {
     177            0 :                             break;
     178           29 :                         }
     179           29 :                         if !self.is_ready(i) {
     180           14 :                             break;
     181           15 :                         }
     182           15 :                         coalesced_ops.push(op);
     183           15 :                         op = self.queued_operations.remove(i).expect("i can't disappear");
     184              :                     }
     185         7077 :                 }
     186              : 
     187        10126 :                 return Some((op, coalesced_ops));
     188        41735 :             }
     189              : 
     190              :             // Nothing can bypass a barrier or shutdown. If it wasn't scheduled above, give up.
     191        41735 :             if matches!(candidate, UploadOp::Barrier(_) | UploadOp::Shutdown) {
     192         4809 :                 return None;
     193        36926 :             }
     194        36926 : 
     195        36926 :             // If upload queue reordering is disabled, bail out after the first operation.
     196        36926 :             if *DISABLE_UPLOAD_QUEUE_REORDERING {
     197            0 :                 return None;
     198        36926 :             }
     199              :         }
     200        11851 :         None
     201        26818 :     }
     202              : 
     203              :     /// Returns true if the queued operation at the given position is ready to be uploaded, i.e. if
     204              :     /// it doesn't conflict with any in-progress or queued operations ahead of it. Operations are
     205              :     /// allowed to skip the queue when it's safe to do so, to increase parallelism.
     206              :     ///
     207              :     /// The position must be valid for the queue size.
     208        51910 :     fn is_ready(&self, pos: usize) -> bool {
     209        51910 :         let candidate = self.queued_operations.get(pos).expect("invalid position");
     210        51910 :         self
     211        51910 :             // Look at in-progress operations, in random order.
     212        51910 :             .inprogress_tasks
     213        51910 :             .values()
     214      2334887 :             .map(|task| &task.op)
     215        51910 :             // Then queued operations ahead of the candidate, front-to-back.
     216        51910 :             .chain(self.queued_operations.iter().take(pos))
     217        51910 :             // Keep track of the active index ahead of each operation. This is used to ensure that
     218        51910 :             // an upload doesn't skip the queue too far, such that it modifies a layer that's
     219        51910 :             // referenced by an active index.
     220        51910 :             //
     221        51910 :             // It's okay that in-progress operations are emitted in random order above, since at
     222        51910 :             // most one of them can be an index upload (enforced by can_bypass).
     223      2363634 :             .scan(&self.clean.0, |next_active_index, op| {
     224      2363634 :                 let active_index = *next_active_index;
     225      2363634 :                 if let UploadOp::UploadMetadata { uploaded } = op {
     226        32782 :                     *next_active_index = uploaded; // stash index for next operation after this
     227      2330852 :                 }
     228      2363634 :                 Some((op, active_index))
     229      2363634 :             })
     230        51910 :             // Check if the candidate can bypass all of them.
     231      2363634 :             .all(|(op, active_index)| candidate.can_bypass(op, active_index))
     232        51910 :     }
     233              : 
     234              :     /// Returns the number of in-progress deletion operations.
     235              :     #[cfg(test)]
     236            4 :     pub(crate) fn num_inprogress_deletions(&self) -> usize {
     237            4 :         self.inprogress_tasks
     238            4 :             .iter()
     239            4 :             .filter(|(_, t)| matches!(t.op, UploadOp::Delete(_)))
     240            4 :             .count()
     241            4 :     }
     242              : 
     243              :     /// Returns the number of in-progress layer uploads.
     244              :     #[cfg(test)]
     245            8 :     pub(crate) fn num_inprogress_layer_uploads(&self) -> usize {
     246            8 :         self.inprogress_tasks
     247            8 :             .iter()
     248           12 :             .filter(|(_, t)| matches!(t.op, UploadOp::UploadLayer(_, _, _)))
     249            8 :             .count()
     250            8 :     }
     251              : 
     252              :     /// Test helper that schedules all ready operations into inprogress_tasks, and returns
     253              :     /// references to them.
     254              :     ///
     255              :     /// TODO: the corresponding production logic should be moved from RemoteTimelineClient into
     256              :     /// UploadQueue, so we can use the same code path.
     257              :     #[cfg(test)]
     258          156 :     fn schedule_ready(&mut self) -> Vec<Arc<UploadTask>> {
     259          156 :         let mut tasks = Vec::new();
     260              :         // NB: schedule operations one by one, to handle conflicts with inprogress_tasks.
     261          344 :         while let Some((op, coalesced_ops)) = self.next_ready() {
     262          188 :             self.task_counter += 1;
     263          188 :             let task = Arc::new(UploadTask {
     264          188 :                 task_id: self.task_counter,
     265          188 :                 op,
     266          188 :                 coalesced_ops,
     267          188 :                 retries: 0.into(),
     268          188 :             });
     269          188 :             self.inprogress_tasks.insert(task.task_id, task.clone());
     270          188 :             tasks.push(task);
     271          188 :         }
     272          156 :         tasks
     273          156 :     }
     274              : 
     275              :     /// Test helper that marks an operation as completed, removing it from inprogress_tasks.
     276              :     ///
     277              :     /// TODO: the corresponding production logic should be moved from RemoteTimelineClient into
     278              :     /// UploadQueue, so we can use the same code path.
     279              :     #[cfg(test)]
     280          116 :     fn complete(&mut self, task_id: u64) {
     281          116 :         let Some(task) = self.inprogress_tasks.remove(&task_id) else {
     282            0 :             return;
     283              :         };
     284              :         // Update the clean index on uploads.
     285          116 :         if let UploadOp::UploadMetadata { ref uploaded } = task.op {
     286           32 :             if task.task_id > self.clean.1.unwrap_or_default() {
     287           32 :                 self.clean = (*uploaded.clone(), Some(task.task_id));
     288           32 :             }
     289           84 :         }
     290          116 :     }
     291              : }
     292              : 
     293              : #[derive(Clone, Copy)]
     294              : pub(super) enum SetDeletedFlagProgress {
     295              :     NotRunning,
     296              :     InProgress(NaiveDateTime),
     297              :     Successful(NaiveDateTime),
     298              : }
     299              : 
     300              : pub struct UploadQueueStoppedDeletable {
     301              :     pub(super) upload_queue_for_deletion: UploadQueueInitialized,
     302              :     pub(super) deleted_at: SetDeletedFlagProgress,
     303              : }
     304              : 
     305              : pub enum UploadQueueStopped {
     306              :     Deletable(UploadQueueStoppedDeletable),
     307              :     Uninitialized,
     308              : }
     309              : 
     310              : #[derive(thiserror::Error, Debug)]
     311              : pub enum NotInitialized {
     312              :     #[error("queue is in state Uninitialized")]
     313              :     Uninitialized,
     314              :     #[error("queue is in state Stopped")]
     315              :     Stopped,
     316              :     #[error("queue is shutting down")]
     317              :     ShuttingDown,
     318              : }
     319              : 
     320              : impl NotInitialized {
     321            0 :     pub(crate) fn is_stopping(&self) -> bool {
     322              :         use NotInitialized::*;
     323            0 :         match self {
     324            0 :             Uninitialized => false,
     325            0 :             Stopped => true,
     326            0 :             ShuttingDown => true,
     327              :         }
     328            0 :     }
     329              : }
     330              : 
     331              : impl UploadQueue {
     332          908 :     pub fn initialize_empty_remote(
     333          908 :         &mut self,
     334          908 :         metadata: &TimelineMetadata,
     335          908 :         inprogress_limit: usize,
     336          908 :     ) -> anyhow::Result<&mut UploadQueueInitialized> {
     337          908 :         match self {
     338          908 :             UploadQueue::Uninitialized => (),
     339              :             UploadQueue::Initialized(_) | UploadQueue::Stopped(_) => {
     340            0 :                 anyhow::bail!("already initialized, state {}", self.as_str())
     341              :             }
     342              :         }
     343              : 
     344          908 :         info!("initializing upload queue for empty remote");
     345              : 
     346          908 :         let index_part = IndexPart::empty(metadata.clone());
     347          908 : 
     348          908 :         let state = UploadQueueInitialized {
     349          908 :             inprogress_limit,
     350          908 :             dirty: index_part.clone(),
     351          908 :             clean: (index_part, None),
     352          908 :             latest_files_changes_since_metadata_upload_scheduled: 0,
     353          908 :             visible_remote_consistent_lsn: Arc::new(AtomicLsn::new(0)),
     354          908 :             // what follows are boring default initializations
     355          908 :             task_counter: 0,
     356          908 :             inprogress_tasks: HashMap::new(),
     357          908 :             queued_operations: VecDeque::new(),
     358          908 :             #[cfg(feature = "testing")]
     359          908 :             dangling_files: HashMap::new(),
     360          908 :             recently_deleted: HashSet::new(),
     361          908 :             blocked_deletions: Vec::new(),
     362          908 :             shutting_down: false,
     363          908 :             shutdown_ready: Arc::new(tokio::sync::Semaphore::new(0)),
     364          908 :         };
     365          908 : 
     366          908 :         *self = UploadQueue::Initialized(state);
     367          908 :         Ok(self.initialized_mut().expect("we just set it"))
     368          908 :     }
     369              : 
     370           44 :     pub fn initialize_with_current_remote_index_part(
     371           44 :         &mut self,
     372           44 :         index_part: &IndexPart,
     373           44 :         inprogress_limit: usize,
     374           44 :     ) -> anyhow::Result<&mut UploadQueueInitialized> {
     375           44 :         match self {
     376           44 :             UploadQueue::Uninitialized => (),
     377              :             UploadQueue::Initialized(_) | UploadQueue::Stopped(_) => {
     378            0 :                 anyhow::bail!("already initialized, state {}", self.as_str())
     379              :             }
     380              :         }
     381              : 
     382           44 :         info!(
     383            0 :             "initializing upload queue with remote index_part.disk_consistent_lsn: {}",
     384            0 :             index_part.metadata.disk_consistent_lsn()
     385              :         );
     386              : 
     387           44 :         let state = UploadQueueInitialized {
     388           44 :             inprogress_limit,
     389           44 :             dirty: index_part.clone(),
     390           44 :             clean: (index_part.clone(), None),
     391           44 :             latest_files_changes_since_metadata_upload_scheduled: 0,
     392           44 :             visible_remote_consistent_lsn: Arc::new(
     393           44 :                 index_part.metadata.disk_consistent_lsn().into(),
     394           44 :             ),
     395           44 :             // what follows are boring default initializations
     396           44 :             task_counter: 0,
     397           44 :             inprogress_tasks: HashMap::new(),
     398           44 :             queued_operations: VecDeque::new(),
     399           44 :             #[cfg(feature = "testing")]
     400           44 :             dangling_files: HashMap::new(),
     401           44 :             recently_deleted: HashSet::new(),
     402           44 :             blocked_deletions: Vec::new(),
     403           44 :             shutting_down: false,
     404           44 :             shutdown_ready: Arc::new(tokio::sync::Semaphore::new(0)),
     405           44 :         };
     406           44 : 
     407           44 :         *self = UploadQueue::Initialized(state);
     408           44 :         Ok(self.initialized_mut().expect("we just set it"))
     409           44 :     }
     410              : 
     411        18375 :     pub fn initialized_mut(&mut self) -> Result<&mut UploadQueueInitialized, NotInitialized> {
     412              :         use UploadQueue::*;
     413        18375 :         match self {
     414            0 :             Uninitialized => Err(NotInitialized::Uninitialized),
     415        18375 :             Initialized(x) => {
     416        18375 :                 if x.shutting_down {
     417            0 :                     Err(NotInitialized::ShuttingDown)
     418              :                 } else {
     419        18375 :                     Ok(x)
     420              :                 }
     421              :             }
     422            0 :             Stopped(_) => Err(NotInitialized::Stopped),
     423              :         }
     424        18375 :     }
     425              : 
     426            4 :     pub(crate) fn stopped_mut(&mut self) -> anyhow::Result<&mut UploadQueueStoppedDeletable> {
     427            4 :         match self {
     428              :             UploadQueue::Initialized(_) | UploadQueue::Uninitialized => {
     429            0 :                 anyhow::bail!("queue is in state {}", self.as_str())
     430              :             }
     431              :             UploadQueue::Stopped(UploadQueueStopped::Uninitialized) => {
     432            0 :                 anyhow::bail!("queue is in state Stopped(Uninitialized)")
     433              :             }
     434            4 :             UploadQueue::Stopped(UploadQueueStopped::Deletable(deletable)) => Ok(deletable),
     435              :         }
     436            4 :     }
     437              : }
     438              : 
     439              : /// An in-progress upload or delete task.
     440              : #[derive(Debug)]
     441              : pub struct UploadTask {
     442              :     /// Unique ID of this task. Used as the key in `inprogress_tasks` above.
     443              :     pub task_id: u64,
     444              :     /// Number of task retries.
     445              :     pub retries: AtomicU32,
     446              :     /// The upload operation.
     447              :     pub op: UploadOp,
     448              :     /// Any upload operations that were coalesced into this operation. This typically happens with
     449              :     /// back-to-back index uploads, see `UploadQueueInitialized::next_ready()`.
     450              :     pub coalesced_ops: Vec<UploadOp>,
     451              : }
     452              : 
     453              : /// A deletion of some layers within the lifetime of a timeline.  This is not used
     454              : /// for timeline deletion, which skips this queue and goes directly to DeletionQueue.
     455              : #[derive(Debug, Clone)]
     456              : pub struct Delete {
     457              :     pub layers: Vec<(LayerName, LayerFileMetadata)>,
     458              : }
     459              : 
     460              : #[derive(Clone, Debug)]
     461              : pub enum UploadOp {
     462              :     /// Upload a layer file. The last field indicates the last operation for thie file.
     463              :     UploadLayer(ResidentLayer, LayerFileMetadata, Option<OpType>),
     464              : 
     465              :     /// Upload a index_part.json file
     466              :     UploadMetadata {
     467              :         /// The next [`UploadQueueInitialized::clean`] after this upload succeeds.
     468              :         uploaded: Box<IndexPart>,
     469              :     },
     470              : 
     471              :     /// Delete layer files
     472              :     Delete(Delete),
     473              : 
     474              :     /// Barrier. When the barrier operation is reached, the channel is closed.
     475              :     Barrier(tokio::sync::watch::Sender<()>),
     476              : 
     477              :     /// Shutdown; upon encountering this operation no new operations will be spawned, otherwise
     478              :     /// this is the same as a Barrier.
     479              :     Shutdown,
     480              : }
     481              : 
     482              : impl std::fmt::Display for UploadOp {
     483            0 :     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
     484            0 :         match self {
     485            0 :             UploadOp::UploadLayer(layer, metadata, mode) => {
     486            0 :                 write!(
     487            0 :                     f,
     488            0 :                     "UploadLayer({}, size={:?}, gen={:?}, mode={:?})",
     489            0 :                     layer, metadata.file_size, metadata.generation, mode
     490            0 :                 )
     491              :             }
     492            0 :             UploadOp::UploadMetadata { uploaded, .. } => {
     493            0 :                 write!(
     494            0 :                     f,
     495            0 :                     "UploadMetadata(lsn: {})",
     496            0 :                     uploaded.metadata.disk_consistent_lsn()
     497            0 :                 )
     498              :             }
     499            0 :             UploadOp::Delete(delete) => {
     500            0 :                 write!(f, "Delete({} layers)", delete.layers.len())
     501              :             }
     502            0 :             UploadOp::Barrier(_) => write!(f, "Barrier"),
     503            0 :             UploadOp::Shutdown => write!(f, "Shutdown"),
     504              :         }
     505            0 :     }
     506              : }
     507              : 
     508              : impl UploadOp {
     509              :     /// Returns true if self can bypass other, i.e. if the operations don't conflict. index is the
     510              :     /// active index when other would be uploaded -- if we allow self to bypass other, this would
     511              :     /// be the active index when self is uploaded.
     512      2363730 :     pub fn can_bypass(&self, other: &UploadOp, index: &IndexPart) -> bool {
     513      2363730 :         match (self, other) {
     514              :             // Nothing can bypass a barrier or shutdown, and it can't bypass anything.
     515         4825 :             (UploadOp::Barrier(_), _) | (_, UploadOp::Barrier(_)) => false,
     516           16 :             (UploadOp::Shutdown, _) | (_, UploadOp::Shutdown) => false,
     517              : 
     518              :             // Uploads and deletes can bypass each other unless they're for the same file.
     519        40069 :             (UploadOp::UploadLayer(a, ameta, _), UploadOp::UploadLayer(b, bmeta, _)) => {
     520        40069 :                 let aname = &a.layer_desc().layer_name();
     521        40069 :                 let bname = &b.layer_desc().layer_name();
     522        40069 :                 !is_same_remote_layer_path(aname, ameta, bname, bmeta)
     523              :             }
     524          122 :             (UploadOp::UploadLayer(u, umeta, _), UploadOp::Delete(d))
     525      2275595 :             | (UploadOp::Delete(d), UploadOp::UploadLayer(u, umeta, _)) => {
     526      2275745 :                 d.layers.iter().all(|(dname, dmeta)| {
     527      2275745 :                     !is_same_remote_layer_path(&u.layer_desc().layer_name(), umeta, dname, dmeta)
     528      2275745 :                 })
     529              :             }
     530              : 
     531              :             // Deletes are idempotent and can always bypass each other.
     532         5693 :             (UploadOp::Delete(_), UploadOp::Delete(_)) => true,
     533              : 
     534              :             // Uploads and deletes can bypass an index upload as long as neither the uploaded index
     535              :             // nor the active index below it references the file. A layer can't be modified or
     536              :             // deleted while referenced by an index.
     537              :             //
     538              :             // Similarly, index uploads can bypass uploads and deletes as long as neither the
     539              :             // uploaded index nor the active index references the file (the latter would be
     540              :             // incorrect use by the caller).
     541          185 :             (UploadOp::UploadLayer(u, umeta, _), UploadOp::UploadMetadata { uploaded: i })
     542         6895 :             | (UploadOp::UploadMetadata { uploaded: i }, UploadOp::UploadLayer(u, umeta, _)) => {
     543         7080 :                 let uname = u.layer_desc().layer_name();
     544         7080 :                 !i.references(&uname, umeta) && !index.references(&uname, umeta)
     545              :             }
     546        29944 :             (UploadOp::Delete(d), UploadOp::UploadMetadata { uploaded: i })
     547           88 :             | (UploadOp::UploadMetadata { uploaded: i }, UploadOp::Delete(d)) => {
     548        30032 :                 d.layers.iter().all(|(dname, dmeta)| {
     549        30032 :                     !i.references(dname, dmeta) && !index.references(dname, dmeta)
     550        30032 :                 })
     551              :             }
     552              : 
     553              :             // Indexes can never bypass each other. They can coalesce though, and
     554              :             // `UploadQueue::next_ready()` currently does this when possible.
     555          298 :             (UploadOp::UploadMetadata { .. }, UploadOp::UploadMetadata { .. }) => false,
     556              :         }
     557      2363730 :     }
     558              : }
     559              : 
     560              : #[cfg(test)]
     561              : mod tests {
     562              :     use std::str::FromStr as _;
     563              : 
     564              :     use itertools::Itertools as _;
     565              :     use utils::shard::{ShardCount, ShardIndex, ShardNumber};
     566              : 
     567              :     use super::*;
     568              :     use crate::DEFAULT_PG_VERSION;
     569              :     use crate::tenant::Timeline;
     570              :     use crate::tenant::harness::{TIMELINE_ID, TenantHarness};
     571              :     use crate::tenant::storage_layer::Layer;
     572              :     use crate::tenant::storage_layer::layer::local_layer_path;
     573              : 
     574              :     /// Test helper which asserts that two operations are the same, in lieu of UploadOp PartialEq.
     575              :     #[track_caller]
     576          196 :     fn assert_same_op(a: &UploadOp, b: &UploadOp) {
     577              :         use UploadOp::*;
     578          196 :         match (a, b) {
     579           88 :             (UploadLayer(a, ameta, atype), UploadLayer(b, bmeta, btype)) => {
     580           88 :                 assert_eq!(a.layer_desc().layer_name(), b.layer_desc().layer_name());
     581           88 :                 assert_eq!(ameta, bmeta);
     582           88 :                 assert_eq!(atype, btype);
     583              :             }
     584           44 :             (Delete(a), Delete(b)) => assert_eq!(a.layers, b.layers),
     585           56 :             (UploadMetadata { uploaded: a }, UploadMetadata { uploaded: b }) => assert_eq!(a, b),
     586            8 :             (Barrier(_), Barrier(_)) => {}
     587            0 :             (Shutdown, Shutdown) => {}
     588            0 :             (a, b) => panic!("{a:?} != {b:?}"),
     589              :         }
     590          196 :     }
     591              : 
     592              :     /// Test helper which asserts that two sets of operations are the same.
     593              :     #[track_caller]
     594           44 :     fn assert_same_ops<'a>(
     595           44 :         a: impl IntoIterator<Item = &'a UploadOp>,
     596           44 :         b: impl IntoIterator<Item = &'a UploadOp>,
     597           44 :     ) {
     598           44 :         a.into_iter()
     599           44 :             .zip_eq(b)
     600          116 :             .for_each(|(a, b)| assert_same_op(a, b))
     601           44 :     }
     602              : 
     603              :     /// Test helper to construct a test timeline.
     604              :     ///
     605              :     /// TODO: it really shouldn't be necessary to construct an entire tenant and timeline just to
     606              :     /// test the upload queue -- decouple ResidentLayer from Timeline.
     607              :     ///
     608              :     /// TODO: the upload queue uses TimelineMetadata::example() instead, because there's no way to
     609              :     /// obtain a TimelineMetadata from a Timeline.
     610           48 :     fn make_timeline() -> Arc<Timeline> {
     611           48 :         // Grab the current test name from the current thread name.
     612           48 :         // TODO: TenantHarness shouldn't take a &'static str, but just leak the test name for now.
     613           48 :         let test_name = std::thread::current().name().unwrap().to_string();
     614           48 :         let test_name = Box::leak(test_name.into_boxed_str());
     615           48 : 
     616           48 :         let runtime = tokio::runtime::Builder::new_current_thread()
     617           48 :             .enable_all()
     618           48 :             .build()
     619           48 :             .expect("failed to create runtime");
     620           48 : 
     621           48 :         runtime
     622           48 :             .block_on(async {
     623           48 :                 let harness = TenantHarness::create(test_name).await?;
     624           48 :                 let (tenant, ctx) = harness.load().await;
     625           48 :                 tenant
     626           48 :                     .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)
     627           48 :                     .await
     628           48 :             })
     629           48 :             .expect("failed to create timeline")
     630           48 :     }
     631              : 
     632              :     /// Test helper to construct an (empty) resident layer.
     633          120 :     fn make_layer(timeline: &Arc<Timeline>, name: &str) -> ResidentLayer {
     634          120 :         make_layer_with_size(timeline, name, 0)
     635          120 :     }
     636              : 
     637              :     /// Test helper to construct a resident layer with the given size.
     638          132 :     fn make_layer_with_size(timeline: &Arc<Timeline>, name: &str, size: usize) -> ResidentLayer {
     639          132 :         let metadata = LayerFileMetadata {
     640          132 :             generation: timeline.generation,
     641          132 :             shard: timeline.get_shard_index(),
     642          132 :             file_size: size as u64,
     643          132 :         };
     644          132 :         make_layer_with_metadata(timeline, name, metadata)
     645          132 :     }
     646              : 
     647              :     /// Test helper to construct a layer with the given metadata.
     648          196 :     fn make_layer_with_metadata(
     649          196 :         timeline: &Arc<Timeline>,
     650          196 :         name: &str,
     651          196 :         metadata: LayerFileMetadata,
     652          196 :     ) -> ResidentLayer {
     653          196 :         let name = LayerName::from_str(name).expect("invalid name");
     654          196 :         let local_path = local_layer_path(
     655          196 :             timeline.conf,
     656          196 :             &timeline.tenant_shard_id,
     657          196 :             &timeline.timeline_id,
     658          196 :             &name,
     659          196 :             &metadata.generation,
     660          196 :         );
     661          196 :         std::fs::write(&local_path, vec![0; metadata.file_size as usize])
     662          196 :             .expect("failed to write file");
     663          196 :         Layer::for_resident(timeline.conf, timeline, local_path, name, metadata)
     664          196 :     }
     665              : 
     666              :     /// Test helper to add a layer to an index and return a new index.
     667           24 :     fn index_with(index: &IndexPart, layer: &ResidentLayer) -> Box<IndexPart> {
     668           24 :         let mut index = index.clone();
     669           24 :         index
     670           24 :             .layer_metadata
     671           24 :             .insert(layer.layer_desc().layer_name(), layer.metadata());
     672           24 :         Box::new(index)
     673           24 :     }
     674              : 
     675              :     /// Test helper to remove a layer from an index and return a new index.
     676            8 :     fn index_without(index: &IndexPart, layer: &ResidentLayer) -> Box<IndexPart> {
     677            8 :         let mut index = index.clone();
     678            8 :         index
     679            8 :             .layer_metadata
     680            8 :             .remove(&layer.layer_desc().layer_name());
     681            8 :         Box::new(index)
     682            8 :     }
     683              : 
     684              :     /// Nothing can bypass a barrier, and it can't bypass inprogress tasks.
     685              :     #[test]
     686            4 :     fn schedule_barrier() -> anyhow::Result<()> {
     687            4 :         let mut queue = UploadQueue::Uninitialized;
     688            4 :         let queue = queue.initialize_empty_remote(&TimelineMetadata::example(), 0)?;
     689            4 :         let tli = make_timeline();
     690            4 : 
     691            4 :         let index = Box::new(queue.clean.0.clone()); // empty, doesn't matter
     692            4 :         let layer0 = make_layer(
     693            4 :             &tli,
     694            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     695            4 :         );
     696            4 :         let layer1 = make_layer(
     697            4 :             &tli,
     698            4 :             "100000000000000000000000000000000000-200000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     699            4 :         );
     700            4 :         let layer2 = make_layer(
     701            4 :             &tli,
     702            4 :             "200000000000000000000000000000000000-300000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     703            4 :         );
     704            4 :         let layer3 = make_layer(
     705            4 :             &tli,
     706            4 :             "300000000000000000000000000000000000-400000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     707            4 :         );
     708            4 :         let (barrier, _) = tokio::sync::watch::channel(());
     709            4 : 
     710            4 :         // Enqueue non-conflicting upload, delete, and index before and after a barrier.
     711            4 :         let ops = [
     712            4 :             UploadOp::UploadLayer(layer0.clone(), layer0.metadata(), None),
     713            4 :             UploadOp::Delete(Delete {
     714            4 :                 layers: vec![(layer1.layer_desc().layer_name(), layer1.metadata())],
     715            4 :             }),
     716            4 :             UploadOp::UploadMetadata {
     717            4 :                 uploaded: index.clone(),
     718            4 :             },
     719            4 :             UploadOp::Barrier(barrier),
     720            4 :             UploadOp::UploadLayer(layer2.clone(), layer2.metadata(), None),
     721            4 :             UploadOp::Delete(Delete {
     722            4 :                 layers: vec![(layer3.layer_desc().layer_name(), layer3.metadata())],
     723            4 :             }),
     724            4 :             UploadOp::UploadMetadata {
     725            4 :                 uploaded: index.clone(),
     726            4 :             },
     727            4 :         ];
     728            4 : 
     729            4 :         queue.queued_operations.extend(ops.clone());
     730            4 : 
     731            4 :         // Schedule the initial operations ahead of the barrier.
     732            4 :         let tasks = queue.schedule_ready();
     733            4 : 
     734           12 :         assert_same_ops(tasks.iter().map(|t| &t.op), &ops[0..3]);
     735            4 :         assert!(matches!(
     736            4 :             queue.queued_operations.front(),
     737              :             Some(&UploadOp::Barrier(_))
     738              :         ));
     739              : 
     740              :         // Complete the initial operations. The barrier isn't scheduled while they're pending.
     741           16 :         for task in tasks {
     742           12 :             assert!(queue.schedule_ready().is_empty());
     743           12 :             queue.complete(task.task_id);
     744              :         }
     745              : 
     746              :         // Schedule the barrier. The later tasks won't schedule until it completes.
     747            4 :         let tasks = queue.schedule_ready();
     748            4 : 
     749            4 :         assert_eq!(tasks.len(), 1);
     750            4 :         assert!(matches!(tasks[0].op, UploadOp::Barrier(_)));
     751            4 :         assert_eq!(queue.queued_operations.len(), 3);
     752              : 
     753              :         // Complete the barrier. The rest of the tasks schedule immediately.
     754            4 :         queue.complete(tasks[0].task_id);
     755            4 : 
     756            4 :         let tasks = queue.schedule_ready();
     757           12 :         assert_same_ops(tasks.iter().map(|t| &t.op), &ops[4..]);
     758            4 :         assert!(queue.queued_operations.is_empty());
     759              : 
     760            4 :         Ok(())
     761            4 :     }
     762              : 
     763              :     /// Deletes can be scheduled in parallel, even if they're for the same file.
     764              :     #[test]
     765            4 :     fn schedule_delete_parallel() -> anyhow::Result<()> {
     766            4 :         let mut queue = UploadQueue::Uninitialized;
     767            4 :         let queue = queue.initialize_empty_remote(&TimelineMetadata::example(), 0)?;
     768            4 :         let tli = make_timeline();
     769            4 : 
     770            4 :         // Enqueue a bunch of deletes, some with conflicting names.
     771            4 :         let layer0 = make_layer(
     772            4 :             &tli,
     773            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     774            4 :         );
     775            4 :         let layer1 = make_layer(
     776            4 :             &tli,
     777            4 :             "100000000000000000000000000000000000-200000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     778            4 :         );
     779            4 :         let layer2 = make_layer(
     780            4 :             &tli,
     781            4 :             "200000000000000000000000000000000000-300000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     782            4 :         );
     783            4 :         let layer3 = make_layer(
     784            4 :             &tli,
     785            4 :             "300000000000000000000000000000000000-400000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     786            4 :         );
     787            4 : 
     788            4 :         let ops = [
     789            4 :             UploadOp::Delete(Delete {
     790            4 :                 layers: vec![(layer0.layer_desc().layer_name(), layer0.metadata())],
     791            4 :             }),
     792            4 :             UploadOp::Delete(Delete {
     793            4 :                 layers: vec![(layer1.layer_desc().layer_name(), layer1.metadata())],
     794            4 :             }),
     795            4 :             UploadOp::Delete(Delete {
     796            4 :                 layers: vec![
     797            4 :                     (layer1.layer_desc().layer_name(), layer1.metadata()),
     798            4 :                     (layer2.layer_desc().layer_name(), layer2.metadata()),
     799            4 :                 ],
     800            4 :             }),
     801            4 :             UploadOp::Delete(Delete {
     802            4 :                 layers: vec![(layer2.layer_desc().layer_name(), layer2.metadata())],
     803            4 :             }),
     804            4 :             UploadOp::Delete(Delete {
     805            4 :                 layers: vec![(layer3.layer_desc().layer_name(), layer3.metadata())],
     806            4 :             }),
     807            4 :         ];
     808            4 : 
     809            4 :         queue.queued_operations.extend(ops.clone());
     810            4 : 
     811            4 :         // Schedule all ready operations. Since deletes don't conflict, they're all scheduled.
     812            4 :         let tasks = queue.schedule_ready();
     813            4 : 
     814           20 :         assert_same_ops(tasks.iter().map(|t| &t.op), &ops);
     815            4 :         assert!(queue.queued_operations.is_empty());
     816              : 
     817            4 :         Ok(())
     818            4 :     }
     819              : 
     820              :     /// Conflicting uploads are serialized.
     821              :     #[test]
     822            4 :     fn schedule_upload_conflicts() -> anyhow::Result<()> {
     823            4 :         let mut queue = UploadQueue::Uninitialized;
     824            4 :         let queue = queue.initialize_with_current_remote_index_part(&IndexPart::example(), 0)?;
     825            4 :         let tli = make_timeline();
     826            4 : 
     827            4 :         // Enqueue three versions of the same layer, with different file sizes.
     828            4 :         let layer0a = make_layer_with_size(
     829            4 :             &tli,
     830            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     831            4 :             1,
     832            4 :         );
     833            4 :         let layer0b = make_layer_with_size(
     834            4 :             &tli,
     835            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     836            4 :             2,
     837            4 :         );
     838            4 :         let layer0c = make_layer_with_size(
     839            4 :             &tli,
     840            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     841            4 :             3,
     842            4 :         );
     843            4 : 
     844            4 :         let ops = [
     845            4 :             UploadOp::UploadLayer(layer0a.clone(), layer0a.metadata(), None),
     846            4 :             UploadOp::UploadLayer(layer0b.clone(), layer0b.metadata(), None),
     847            4 :             UploadOp::UploadLayer(layer0c.clone(), layer0c.metadata(), None),
     848            4 :         ];
     849            4 : 
     850            4 :         queue.queued_operations.extend(ops.clone());
     851              : 
     852              :         // Only one version should be scheduled and uploaded at a time.
     853           16 :         for op in ops {
     854           12 :             let tasks = queue.schedule_ready();
     855           12 :             assert_eq!(tasks.len(), 1);
     856           12 :             assert_same_op(&tasks[0].op, &op);
     857           12 :             queue.complete(tasks[0].task_id);
     858              :         }
     859            4 :         assert!(queue.schedule_ready().is_empty());
     860            4 :         assert!(queue.queued_operations.is_empty());
     861              : 
     862            4 :         Ok(())
     863            4 :     }
     864              : 
     865              :     /// Conflicting uploads and deletes are serialized.
     866              :     #[test]
     867            4 :     fn schedule_upload_delete_conflicts() -> anyhow::Result<()> {
     868            4 :         let mut queue = UploadQueue::Uninitialized;
     869            4 :         let queue = queue.initialize_with_current_remote_index_part(&IndexPart::example(), 0)?;
     870            4 :         let tli = make_timeline();
     871            4 : 
     872            4 :         // Enqueue two layer uploads, with a delete of both layers in between them. These should be
     873            4 :         // scheduled one at a time, since deletes can't bypass uploads and vice versa.
     874            4 :         let layer0 = make_layer(
     875            4 :             &tli,
     876            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     877            4 :         );
     878            4 :         let layer1 = make_layer(
     879            4 :             &tli,
     880            4 :             "100000000000000000000000000000000000-200000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     881            4 :         );
     882            4 : 
     883            4 :         let ops = [
     884            4 :             UploadOp::UploadLayer(layer0.clone(), layer0.metadata(), None),
     885            4 :             UploadOp::Delete(Delete {
     886            4 :                 layers: vec![
     887            4 :                     (layer0.layer_desc().layer_name(), layer0.metadata()),
     888            4 :                     (layer1.layer_desc().layer_name(), layer1.metadata()),
     889            4 :                 ],
     890            4 :             }),
     891            4 :             UploadOp::UploadLayer(layer1.clone(), layer1.metadata(), None),
     892            4 :         ];
     893            4 : 
     894            4 :         queue.queued_operations.extend(ops.clone());
     895              : 
     896              :         // Only one version should be scheduled and uploaded at a time.
     897           16 :         for op in ops {
     898           12 :             let tasks = queue.schedule_ready();
     899           12 :             assert_eq!(tasks.len(), 1);
     900           12 :             assert_same_op(&tasks[0].op, &op);
     901           12 :             queue.complete(tasks[0].task_id);
     902              :         }
     903            4 :         assert!(queue.schedule_ready().is_empty());
     904            4 :         assert!(queue.queued_operations.is_empty());
     905              : 
     906            4 :         Ok(())
     907            4 :     }
     908              : 
     909              :     /// Non-conflicting uploads and deletes can bypass the queue, avoiding the conflicting
     910              :     /// delete/upload operations at the head of the queue.
     911              :     #[test]
     912            4 :     fn schedule_upload_delete_conflicts_bypass() -> anyhow::Result<()> {
     913            4 :         let mut queue = UploadQueue::Uninitialized;
     914            4 :         let queue = queue.initialize_with_current_remote_index_part(&IndexPart::example(), 0)?;
     915            4 :         let tli = make_timeline();
     916            4 : 
     917            4 :         // Enqueue two layer uploads, with a delete of both layers in between them. These should be
     918            4 :         // scheduled one at a time, since deletes can't bypass uploads and vice versa.
     919            4 :         //
     920            4 :         // Also enqueue non-conflicting uploads and deletes at the end. These can bypass the queue
     921            4 :         // and run immediately.
     922            4 :         let layer0 = make_layer(
     923            4 :             &tli,
     924            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     925            4 :         );
     926            4 :         let layer1 = make_layer(
     927            4 :             &tli,
     928            4 :             "100000000000000000000000000000000000-200000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     929            4 :         );
     930            4 :         let layer2 = make_layer(
     931            4 :             &tli,
     932            4 :             "200000000000000000000000000000000000-300000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     933            4 :         );
     934            4 :         let layer3 = make_layer(
     935            4 :             &tli,
     936            4 :             "300000000000000000000000000000000000-400000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     937            4 :         );
     938            4 : 
     939            4 :         let ops = [
     940            4 :             UploadOp::UploadLayer(layer0.clone(), layer0.metadata(), None),
     941            4 :             UploadOp::Delete(Delete {
     942            4 :                 layers: vec![
     943            4 :                     (layer0.layer_desc().layer_name(), layer0.metadata()),
     944            4 :                     (layer1.layer_desc().layer_name(), layer1.metadata()),
     945            4 :                 ],
     946            4 :             }),
     947            4 :             UploadOp::UploadLayer(layer1.clone(), layer1.metadata(), None),
     948            4 :             UploadOp::UploadLayer(layer2.clone(), layer2.metadata(), None),
     949            4 :             UploadOp::Delete(Delete {
     950            4 :                 layers: vec![(layer3.layer_desc().layer_name(), layer3.metadata())],
     951            4 :             }),
     952            4 :         ];
     953            4 : 
     954            4 :         queue.queued_operations.extend(ops.clone());
     955            4 : 
     956            4 :         // Operations 0, 3, and 4 are scheduled immediately.
     957            4 :         let tasks = queue.schedule_ready();
     958           12 :         assert_same_ops(tasks.iter().map(|t| &t.op), [&ops[0], &ops[3], &ops[4]]);
     959            4 :         assert_eq!(queue.queued_operations.len(), 2);
     960              : 
     961            4 :         Ok(())
     962            4 :     }
     963              : 
     964              :     /// Non-conflicting uploads are parallelized.
     965              :     #[test]
     966            4 :     fn schedule_upload_parallel() -> anyhow::Result<()> {
     967            4 :         let mut queue = UploadQueue::Uninitialized;
     968            4 :         let queue = queue.initialize_with_current_remote_index_part(&IndexPart::example(), 0)?;
     969            4 :         let tli = make_timeline();
     970            4 : 
     971            4 :         // Enqueue three different layer uploads.
     972            4 :         let layer0 = make_layer(
     973            4 :             &tli,
     974            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     975            4 :         );
     976            4 :         let layer1 = make_layer(
     977            4 :             &tli,
     978            4 :             "100000000000000000000000000000000000-200000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     979            4 :         );
     980            4 :         let layer2 = make_layer(
     981            4 :             &tli,
     982            4 :             "200000000000000000000000000000000000-300000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
     983            4 :         );
     984            4 : 
     985            4 :         let ops = [
     986            4 :             UploadOp::UploadLayer(layer0.clone(), layer0.metadata(), None),
     987            4 :             UploadOp::UploadLayer(layer1.clone(), layer1.metadata(), None),
     988            4 :             UploadOp::UploadLayer(layer2.clone(), layer2.metadata(), None),
     989            4 :         ];
     990            4 : 
     991            4 :         queue.queued_operations.extend(ops.clone());
     992            4 : 
     993            4 :         // All uploads should be scheduled concurrently.
     994            4 :         let tasks = queue.schedule_ready();
     995            4 : 
     996           12 :         assert_same_ops(tasks.iter().map(|t| &t.op), &ops);
     997            4 :         assert!(queue.queued_operations.is_empty());
     998              : 
     999            4 :         Ok(())
    1000            4 :     }
    1001              : 
    1002              :     /// Index uploads are coalesced.
    1003              :     #[test]
    1004            4 :     fn schedule_index_coalesce() -> anyhow::Result<()> {
    1005            4 :         let mut queue = UploadQueue::Uninitialized;
    1006            4 :         let queue = queue.initialize_with_current_remote_index_part(&IndexPart::example(), 0)?;
    1007              : 
    1008              :         // Enqueue three uploads of the current empty index.
    1009            4 :         let index = Box::new(queue.clean.0.clone());
    1010            4 : 
    1011            4 :         let ops = [
    1012            4 :             UploadOp::UploadMetadata {
    1013            4 :                 uploaded: index.clone(),
    1014            4 :             },
    1015            4 :             UploadOp::UploadMetadata {
    1016            4 :                 uploaded: index.clone(),
    1017            4 :             },
    1018            4 :             UploadOp::UploadMetadata {
    1019            4 :                 uploaded: index.clone(),
    1020            4 :             },
    1021            4 :         ];
    1022            4 : 
    1023            4 :         queue.queued_operations.extend(ops.clone());
    1024            4 : 
    1025            4 :         // The index uploads are coalesced into a single operation.
    1026            4 :         let tasks = queue.schedule_ready();
    1027            4 :         assert_eq!(tasks.len(), 1);
    1028            4 :         assert_same_op(&tasks[0].op, &ops[2]);
    1029            4 :         assert_same_ops(&tasks[0].coalesced_ops, &ops[0..2]);
    1030            4 : 
    1031            4 :         assert!(queue.queued_operations.is_empty());
    1032              : 
    1033            4 :         Ok(())
    1034            4 :     }
    1035              : 
    1036              :     /// Chains of upload/index operations lead to parallel layer uploads and serial index uploads.
    1037              :     /// This is the common case with layer flushes.
    1038              :     #[test]
    1039            4 :     fn schedule_index_upload_chain() -> anyhow::Result<()> {
    1040            4 :         let mut queue = UploadQueue::Uninitialized;
    1041            4 :         let queue = queue.initialize_with_current_remote_index_part(&IndexPart::example(), 0)?;
    1042            4 :         let tli = make_timeline();
    1043            4 : 
    1044            4 :         // Enqueue three uploads of the current empty index.
    1045            4 :         let index = Box::new(queue.clean.0.clone());
    1046            4 :         let layer0 = make_layer(
    1047            4 :             &tli,
    1048            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1049            4 :         );
    1050            4 :         let index0 = index_with(&index, &layer0);
    1051            4 :         let layer1 = make_layer(
    1052            4 :             &tli,
    1053            4 :             "100000000000000000000000000000000000-200000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1054            4 :         );
    1055            4 :         let index1 = index_with(&index0, &layer1);
    1056            4 :         let layer2 = make_layer(
    1057            4 :             &tli,
    1058            4 :             "200000000000000000000000000000000000-300000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1059            4 :         );
    1060            4 :         let index2 = index_with(&index1, &layer2);
    1061            4 : 
    1062            4 :         let ops = [
    1063            4 :             UploadOp::UploadLayer(layer0.clone(), layer0.metadata(), None),
    1064            4 :             UploadOp::UploadMetadata {
    1065            4 :                 uploaded: index0.clone(),
    1066            4 :             },
    1067            4 :             UploadOp::UploadLayer(layer1.clone(), layer1.metadata(), None),
    1068            4 :             UploadOp::UploadMetadata {
    1069            4 :                 uploaded: index1.clone(),
    1070            4 :             },
    1071            4 :             UploadOp::UploadLayer(layer2.clone(), layer2.metadata(), None),
    1072            4 :             UploadOp::UploadMetadata {
    1073            4 :                 uploaded: index2.clone(),
    1074            4 :             },
    1075            4 :         ];
    1076            4 : 
    1077            4 :         queue.queued_operations.extend(ops.clone());
    1078            4 : 
    1079            4 :         // The layer uploads should be scheduled immediately. The indexes must wait.
    1080            4 :         let upload_tasks = queue.schedule_ready();
    1081            4 :         assert_same_ops(
    1082           12 :             upload_tasks.iter().map(|t| &t.op),
    1083            4 :             [&ops[0], &ops[2], &ops[4]],
    1084            4 :         );
    1085            4 : 
    1086            4 :         // layer2 completes first. None of the indexes can upload yet.
    1087            4 :         queue.complete(upload_tasks[2].task_id);
    1088            4 :         assert!(queue.schedule_ready().is_empty());
    1089              : 
    1090              :         // layer0 completes. index0 can upload. It completes.
    1091            4 :         queue.complete(upload_tasks[0].task_id);
    1092            4 :         let index_tasks = queue.schedule_ready();
    1093            4 :         assert_eq!(index_tasks.len(), 1);
    1094            4 :         assert_same_op(&index_tasks[0].op, &ops[1]);
    1095            4 :         queue.complete(index_tasks[0].task_id);
    1096            4 : 
    1097            4 :         // layer 1 completes. This unblocks index 1 and 2, which coalesce into
    1098            4 :         // a single upload for index 2.
    1099            4 :         queue.complete(upload_tasks[1].task_id);
    1100            4 : 
    1101            4 :         let index_tasks = queue.schedule_ready();
    1102            4 :         assert_eq!(index_tasks.len(), 1);
    1103            4 :         assert_same_op(&index_tasks[0].op, &ops[5]);
    1104            4 :         assert_same_ops(&index_tasks[0].coalesced_ops, &ops[3..4]);
    1105            4 : 
    1106            4 :         assert!(queue.queued_operations.is_empty());
    1107              : 
    1108            4 :         Ok(())
    1109            4 :     }
    1110              : 
    1111              :     /// A delete can't bypass an index upload if an index ahead of it still references it.
    1112              :     #[test]
    1113            4 :     fn schedule_index_delete_dereferenced() -> anyhow::Result<()> {
    1114            4 :         let mut queue = UploadQueue::Uninitialized;
    1115            4 :         let queue = queue.initialize_with_current_remote_index_part(&IndexPart::example(), 0)?;
    1116            4 :         let tli = make_timeline();
    1117            4 : 
    1118            4 :         // Create a layer to upload.
    1119            4 :         let layer = make_layer(
    1120            4 :             &tli,
    1121            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1122            4 :         );
    1123            4 :         let index_upload = index_with(&queue.clean.0, &layer);
    1124            4 : 
    1125            4 :         // Remove the layer reference in a new index, then delete the layer.
    1126            4 :         let index_deref = index_without(&index_upload, &layer);
    1127            4 : 
    1128            4 :         let ops = [
    1129            4 :             // Initial upload, with a barrier to prevent index coalescing.
    1130            4 :             UploadOp::UploadLayer(layer.clone(), layer.metadata(), None),
    1131            4 :             UploadOp::UploadMetadata {
    1132            4 :                 uploaded: index_upload.clone(),
    1133            4 :             },
    1134            4 :             UploadOp::Barrier(tokio::sync::watch::channel(()).0),
    1135            4 :             // Dereference the layer and delete it.
    1136            4 :             UploadOp::UploadMetadata {
    1137            4 :                 uploaded: index_deref.clone(),
    1138            4 :             },
    1139            4 :             UploadOp::Delete(Delete {
    1140            4 :                 layers: vec![(layer.layer_desc().layer_name(), layer.metadata())],
    1141            4 :             }),
    1142            4 :         ];
    1143            4 : 
    1144            4 :         queue.queued_operations.extend(ops.clone());
    1145              : 
    1146              :         // Operations are serialized.
    1147           24 :         for op in ops {
    1148           20 :             let tasks = queue.schedule_ready();
    1149           20 :             assert_eq!(tasks.len(), 1);
    1150           20 :             assert_same_op(&tasks[0].op, &op);
    1151           20 :             queue.complete(tasks[0].task_id);
    1152              :         }
    1153            4 :         assert!(queue.queued_operations.is_empty());
    1154              : 
    1155            4 :         Ok(())
    1156            4 :     }
    1157              : 
    1158              :     /// An upload with a reused layer name doesn't clobber the previous layer. Specifically, a
    1159              :     /// dereference/upload/reference cycle can't allow the upload to bypass the reference.
    1160              :     #[test]
    1161            4 :     fn schedule_index_upload_dereferenced() -> anyhow::Result<()> {
    1162            4 :         let mut queue = UploadQueue::Uninitialized;
    1163            4 :         let queue = queue.initialize_with_current_remote_index_part(&IndexPart::example(), 0)?;
    1164            4 :         let tli = make_timeline();
    1165            4 : 
    1166            4 :         // Create a layer to upload.
    1167            4 :         let layer = make_layer(
    1168            4 :             &tli,
    1169            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1170            4 :         );
    1171            4 : 
    1172            4 :         // Upload the layer. Then dereference the layer, and upload/reference it again.
    1173            4 :         let index_upload = index_with(&queue.clean.0, &layer);
    1174            4 :         let index_deref = index_without(&index_upload, &layer);
    1175            4 :         let index_ref = index_with(&index_deref, &layer);
    1176            4 : 
    1177            4 :         let ops = [
    1178            4 :             // Initial upload, with a barrier to prevent index coalescing.
    1179            4 :             UploadOp::UploadLayer(layer.clone(), layer.metadata(), None),
    1180            4 :             UploadOp::UploadMetadata {
    1181            4 :                 uploaded: index_upload.clone(),
    1182            4 :             },
    1183            4 :             UploadOp::Barrier(tokio::sync::watch::channel(()).0),
    1184            4 :             // Dereference the layer.
    1185            4 :             UploadOp::UploadMetadata {
    1186            4 :                 uploaded: index_deref.clone(),
    1187            4 :             },
    1188            4 :             // Replace and reference the layer.
    1189            4 :             UploadOp::UploadLayer(layer.clone(), layer.metadata(), None),
    1190            4 :             UploadOp::UploadMetadata {
    1191            4 :                 uploaded: index_ref.clone(),
    1192            4 :             },
    1193            4 :         ];
    1194            4 : 
    1195            4 :         queue.queued_operations.extend(ops.clone());
    1196              : 
    1197              :         // Operations are serialized.
    1198           28 :         for op in ops {
    1199           24 :             let tasks = queue.schedule_ready();
    1200           24 :             assert_eq!(tasks.len(), 1);
    1201           24 :             assert_same_op(&tasks[0].op, &op);
    1202           24 :             queue.complete(tasks[0].task_id);
    1203              :         }
    1204            4 :         assert!(queue.queued_operations.is_empty());
    1205              : 
    1206            4 :         Ok(())
    1207            4 :     }
    1208              : 
    1209              :     /// Nothing can bypass a shutdown, and it waits for inprogress tasks. It's never returned from
    1210              :     /// next_ready(), but is left at the head of the queue.
    1211              :     #[test]
    1212            4 :     fn schedule_shutdown() -> anyhow::Result<()> {
    1213            4 :         let mut queue = UploadQueue::Uninitialized;
    1214            4 :         let queue = queue.initialize_empty_remote(&TimelineMetadata::example(), 0)?;
    1215            4 :         let tli = make_timeline();
    1216            4 : 
    1217            4 :         let index = Box::new(queue.clean.0.clone()); // empty, doesn't matter
    1218            4 :         let layer0 = make_layer(
    1219            4 :             &tli,
    1220            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1221            4 :         );
    1222            4 :         let layer1 = make_layer(
    1223            4 :             &tli,
    1224            4 :             "100000000000000000000000000000000000-200000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1225            4 :         );
    1226            4 :         let layer2 = make_layer(
    1227            4 :             &tli,
    1228            4 :             "200000000000000000000000000000000000-300000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1229            4 :         );
    1230            4 :         let layer3 = make_layer(
    1231            4 :             &tli,
    1232            4 :             "300000000000000000000000000000000000-400000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1233            4 :         );
    1234            4 : 
    1235            4 :         // Enqueue non-conflicting upload, delete, and index before and after a shutdown.
    1236            4 :         let ops = [
    1237            4 :             UploadOp::UploadLayer(layer0.clone(), layer0.metadata(), None),
    1238            4 :             UploadOp::Delete(Delete {
    1239            4 :                 layers: vec![(layer1.layer_desc().layer_name(), layer1.metadata())],
    1240            4 :             }),
    1241            4 :             UploadOp::UploadMetadata {
    1242            4 :                 uploaded: index.clone(),
    1243            4 :             },
    1244            4 :             UploadOp::Shutdown,
    1245            4 :             UploadOp::UploadLayer(layer2.clone(), layer2.metadata(), None),
    1246            4 :             UploadOp::Delete(Delete {
    1247            4 :                 layers: vec![(layer3.layer_desc().layer_name(), layer3.metadata())],
    1248            4 :             }),
    1249            4 :             UploadOp::UploadMetadata {
    1250            4 :                 uploaded: index.clone(),
    1251            4 :             },
    1252            4 :         ];
    1253            4 : 
    1254            4 :         queue.queued_operations.extend(ops.clone());
    1255            4 : 
    1256            4 :         // Schedule the initial operations ahead of the shutdown.
    1257            4 :         let tasks = queue.schedule_ready();
    1258            4 : 
    1259           12 :         assert_same_ops(tasks.iter().map(|t| &t.op), &ops[0..3]);
    1260            4 :         assert!(matches!(
    1261            4 :             queue.queued_operations.front(),
    1262              :             Some(&UploadOp::Shutdown)
    1263              :         ));
    1264              : 
    1265              :         // Complete the initial operations. The shutdown isn't triggered while they're pending.
    1266           16 :         for task in tasks {
    1267           12 :             assert!(queue.schedule_ready().is_empty());
    1268           12 :             queue.complete(task.task_id);
    1269              :         }
    1270              : 
    1271              :         // The shutdown is triggered the next time we try to pull an operation. It isn't returned,
    1272              :         // but is left in the queue.
    1273            4 :         assert!(!queue.shutdown_ready.is_closed());
    1274            4 :         assert!(queue.next_ready().is_none());
    1275            4 :         assert!(queue.shutdown_ready.is_closed());
    1276              : 
    1277            4 :         Ok(())
    1278            4 :     }
    1279              : 
    1280              :     /// Scheduling respects inprogress_limit.
    1281              :     #[test]
    1282            4 :     fn schedule_inprogress_limit() -> anyhow::Result<()> {
    1283            4 :         // Create a queue with inprogress_limit=2.
    1284            4 :         let mut queue = UploadQueue::Uninitialized;
    1285            4 :         let queue = queue.initialize_empty_remote(&TimelineMetadata::example(), 2)?;
    1286            4 :         let tli = make_timeline();
    1287            4 : 
    1288            4 :         // Enqueue a bunch of uploads.
    1289            4 :         let layer0 = make_layer(
    1290            4 :             &tli,
    1291            4 :             "000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1292            4 :         );
    1293            4 :         let layer1 = make_layer(
    1294            4 :             &tli,
    1295            4 :             "100000000000000000000000000000000000-200000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1296            4 :         );
    1297            4 :         let layer2 = make_layer(
    1298            4 :             &tli,
    1299            4 :             "200000000000000000000000000000000000-300000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1300            4 :         );
    1301            4 :         let layer3 = make_layer(
    1302            4 :             &tli,
    1303            4 :             "300000000000000000000000000000000000-400000000000000000000000000000000000__00000000016B59D8-00000000016B5A51",
    1304            4 :         );
    1305            4 : 
    1306            4 :         let ops = [
    1307            4 :             UploadOp::UploadLayer(layer0.clone(), layer0.metadata(), None),
    1308            4 :             UploadOp::UploadLayer(layer1.clone(), layer1.metadata(), None),
    1309            4 :             UploadOp::UploadLayer(layer2.clone(), layer2.metadata(), None),
    1310            4 :             UploadOp::UploadLayer(layer3.clone(), layer3.metadata(), None),
    1311            4 :         ];
    1312            4 : 
    1313            4 :         queue.queued_operations.extend(ops.clone());
    1314            4 : 
    1315            4 :         // Schedule all ready operations. Only 2 are scheduled.
    1316            4 :         let tasks = queue.schedule_ready();
    1317            8 :         assert_same_ops(tasks.iter().map(|t| &t.op), &ops[0..2]);
    1318            4 :         assert!(queue.next_ready().is_none());
    1319              : 
    1320              :         // When one completes, another is scheduled.
    1321            4 :         queue.complete(tasks[0].task_id);
    1322            4 :         let tasks = queue.schedule_ready();
    1323            4 :         assert_same_ops(tasks.iter().map(|t| &t.op), &ops[2..3]);
    1324            4 : 
    1325            4 :         Ok(())
    1326            4 :     }
    1327              : 
    1328              :     /// Tests that can_bypass takes name, generation and shard index into account for all operations.
    1329              :     #[test]
    1330            4 :     fn can_bypass_path() -> anyhow::Result<()> {
    1331            4 :         let tli = make_timeline();
    1332            4 : 
    1333            4 :         let name0 = &"000000000000000000000000000000000000-100000000000000000000000000000000000__00000000016B59D8-00000000016B5A51";
    1334            4 :         let name1 = &"100000000000000000000000000000000000-200000000000000000000000000000000000__00000000016B59D8-00000000016B5A51";
    1335              : 
    1336              :         // Asserts that layers a and b either can or can't bypass each other, for all combinations
    1337              :         // of operations (except Delete and UploadMetadata which are special-cased).
    1338              :         #[track_caller]
    1339           32 :         fn assert_can_bypass(a: ResidentLayer, b: ResidentLayer, can_bypass: bool) {
    1340           32 :             let index = IndexPart::empty(TimelineMetadata::example());
    1341           96 :             for (a, b) in make_ops(a).into_iter().zip(make_ops(b)) {
    1342           96 :                 match (&a, &b) {
    1343              :                     // Deletes can always bypass each other.
    1344           32 :                     (UploadOp::Delete(_), UploadOp::Delete(_)) => assert!(a.can_bypass(&b, &index)),
    1345              :                     // Indexes can never bypass each other.
    1346              :                     (UploadOp::UploadMetadata { .. }, UploadOp::UploadMetadata { .. }) => {
    1347           32 :                         assert!(!a.can_bypass(&b, &index))
    1348              :                     }
    1349              :                     // For other operations, assert as requested.
    1350           32 :                     (a, b) => assert_eq!(a.can_bypass(b, &index), can_bypass),
    1351              :                 }
    1352              :             }
    1353           32 :         }
    1354              : 
    1355           64 :         fn make_ops(layer: ResidentLayer) -> Vec<UploadOp> {
    1356           64 :             let mut index = IndexPart::empty(TimelineMetadata::example());
    1357           64 :             index
    1358           64 :                 .layer_metadata
    1359           64 :                 .insert(layer.layer_desc().layer_name(), layer.metadata());
    1360           64 :             vec![
    1361           64 :                 UploadOp::UploadLayer(layer.clone(), layer.metadata(), None),
    1362           64 :                 UploadOp::Delete(Delete {
    1363           64 :                     layers: vec![(layer.layer_desc().layer_name(), layer.metadata())],
    1364           64 :                 }),
    1365           64 :                 UploadOp::UploadMetadata {
    1366           64 :                     uploaded: Box::new(index),
    1367           64 :                 },
    1368           64 :             ]
    1369           64 :         }
    1370              : 
    1371              :         // Makes a ResidentLayer.
    1372           64 :         let layer = |name: &'static str, shard: Option<u8>, generation: u32| -> ResidentLayer {
    1373           64 :             let shard = shard
    1374           64 :                 .map(|n| ShardIndex::new(ShardNumber(n), ShardCount(8)))
    1375           64 :                 .unwrap_or(ShardIndex::unsharded());
    1376           64 :             let metadata = LayerFileMetadata {
    1377           64 :                 shard,
    1378           64 :                 generation: Generation::Valid(generation),
    1379           64 :                 file_size: 0,
    1380           64 :             };
    1381           64 :             make_layer_with_metadata(&tli, name, metadata)
    1382           64 :         };
    1383              : 
    1384              :         // Same name and metadata can't bypass. This goes both for unsharded and sharded, as well as
    1385              :         // 0 or >0 generation.
    1386            4 :         assert_can_bypass(layer(name0, None, 0), layer(name0, None, 0), false);
    1387            4 :         assert_can_bypass(layer(name0, Some(0), 0), layer(name0, Some(0), 0), false);
    1388            4 :         assert_can_bypass(layer(name0, None, 1), layer(name0, None, 1), false);
    1389            4 : 
    1390            4 :         // Different names can bypass.
    1391            4 :         assert_can_bypass(layer(name0, None, 0), layer(name1, None, 0), true);
    1392            4 : 
    1393            4 :         // Different shards can bypass. Shard 0 is different from unsharded.
    1394            4 :         assert_can_bypass(layer(name0, Some(0), 0), layer(name0, Some(1), 0), true);
    1395            4 :         assert_can_bypass(layer(name0, Some(0), 0), layer(name0, None, 0), true);
    1396            4 : 
    1397            4 :         // Different generations can bypass, both sharded and unsharded.
    1398            4 :         assert_can_bypass(layer(name0, None, 0), layer(name0, None, 1), true);
    1399            4 :         assert_can_bypass(layer(name0, Some(1), 0), layer(name0, Some(1), 1), true);
    1400            4 : 
    1401            4 :         Ok(())
    1402            4 :     }
    1403              : }
        

Generated by: LCOV version 2.1-beta