diff --git a/pageserver/src/deletion_queue.rs b/pageserver/src/deletion_queue.rs index 034ee242e6..eefc436463 100644 --- a/pageserver/src/deletion_queue.rs +++ b/pageserver/src/deletion_queue.rs @@ -13,6 +13,7 @@ use thiserror::Error; use tokio; use tokio_util::sync::CancellationToken; use tracing::{self, debug, error}; +use utils::generation::Generation; use utils::id::{TenantId, TenantTimelineId, TimelineId}; pub(crate) use self::backend::BackendQueueWorker; @@ -204,7 +205,7 @@ impl DeletionQueueClient { &self, tenant_id: TenantId, timeline_id: TimelineId, - layers: Vec, + layers: Vec<(LayerFileName, Generation)>, ) -> Result<(), DeletionQueueError> { DELETION_QUEUE_SUBMITTED.inc_by(layers.len() as u64); self.do_push(FrontendQueueMessage::Delete(DeletionOp { @@ -363,7 +364,7 @@ mod test { use remote_storage::{RemoteStorageConfig, RemoteStorageKind}; use tokio::{runtime::EnterGuard, task::JoinHandle}; - use crate::tenant::harness::TenantHarness; + use crate::tenant::{harness::TenantHarness, remote_timeline_client::remote_timeline_path}; use super::*; pub const TIMELINE_ID: TimelineId = @@ -538,38 +539,37 @@ mod test { let tenant_id = ctx.harness.tenant_id; let content: Vec = "victim1 contents".into(); - let relative_remote_path = ctx - .harness - .conf - .remote_path(&ctx.harness.timeline_path(&TIMELINE_ID)) - .expect("Failed to construct remote path"); + let relative_remote_path = remote_timeline_path(&tenant_id, &TIMELINE_ID); let remote_timeline_path = ctx.remote_fs_dir.join(relative_remote_path.get_path()); let deletion_prefix = ctx.harness.conf.deletion_prefix(); + let generation = Generation::new(0xdeadbeef); + let remote_layer_file_name_1 = format!("{}{}", layer_file_name_1, generation.get_suffix()); + // Inject a victim file to remote storage info!("Writing"); std::fs::create_dir_all(&remote_timeline_path)?; std::fs::write( - remote_timeline_path.join(layer_file_name_1.to_string()), + remote_timeline_path.join(remote_layer_file_name_1.clone()), content, )?; - assert_remote_files(&[&layer_file_name_1.file_name()], &remote_timeline_path); + assert_remote_files(&[&remote_layer_file_name_1], &remote_timeline_path); // File should still be there after we push it to the queue (we haven't pushed enough to flush anything) info!("Pushing"); ctx.runtime.block_on(client.push_layers( tenant_id, TIMELINE_ID, - [layer_file_name_1.clone()].to_vec(), + [(layer_file_name_1.clone(), generation)].to_vec(), ))?; - assert_remote_files(&[&layer_file_name_1.file_name()], &remote_timeline_path); + assert_remote_files(&[&remote_layer_file_name_1], &remote_timeline_path); assert_local_files(&[], &deletion_prefix); // File should still be there after we write a deletion list (we haven't pushed enough to execute anything) info!("Flushing"); ctx.runtime.block_on(client.flush())?; - assert_remote_files(&[&layer_file_name_1.file_name()], &remote_timeline_path); + assert_remote_files(&[&remote_layer_file_name_1], &remote_timeline_path); assert_local_files(&["0000000000000001-01.list"], &deletion_prefix); // File should go away when we execute @@ -596,24 +596,22 @@ mod test { let tenant_id = ctx.harness.tenant_id; let content: Vec = "victim1 contents".into(); - let relative_remote_path = ctx - .harness - .conf - .remote_path(&ctx.harness.timeline_path(&TIMELINE_ID)) - .expect("Failed to construct remote path"); + let relative_remote_path = remote_timeline_path(&tenant_id, &TIMELINE_ID); let remote_timeline_path = ctx.remote_fs_dir.join(relative_remote_path.get_path()); let deletion_prefix = ctx.harness.conf.deletion_prefix(); + let generation = Generation::new(0xdeadbeef); + let remote_layer_file_name_1 = format!("{}{}", layer_file_name_1, generation.get_suffix()); // Inject a file, delete it, and flush to a deletion list std::fs::create_dir_all(&remote_timeline_path)?; std::fs::write( - remote_timeline_path.join(layer_file_name_1.to_string()), + remote_timeline_path.join(remote_layer_file_name_1.clone()), content, )?; ctx.runtime.block_on(client.push_layers( tenant_id, TIMELINE_ID, - [layer_file_name_1.clone()].to_vec(), + [(layer_file_name_1.clone(), generation)].to_vec(), ))?; ctx.runtime.block_on(client.flush())?; assert_local_files(&["0000000000000001-01.list"], &deletion_prefix); @@ -638,6 +636,8 @@ mod test { pub mod mock { use tracing::info; + use crate::tenant::remote_timeline_client::remote_layer_path; + use super::*; use std::sync::{ atomic::{AtomicUsize, Ordering}, @@ -652,10 +652,7 @@ pub mod mock { } impl MockDeletionQueue { - pub fn new( - remote_storage: Option, - conf: &'static PageServerConf, - ) -> Self { + pub fn new(remote_storage: Option) -> Self { let (tx, mut rx) = tokio::sync::mpsc::channel(16384); let (tx_pump, mut rx_pump) = tokio::sync::mpsc::channel::(1); let (executor_tx, mut executor_rx) = tokio::sync::mpsc::channel(16384); @@ -703,19 +700,14 @@ pub mod mock { while let Ok(msg) = rx.try_recv() { match msg { FrontendQueueMessage::Delete(op) => { - let timeline_path = - conf.timeline_path(&op.tenant_id, &op.timeline_id); - let mut objects = op.objects; - for layer in op.layers { - let local_path = timeline_path.join(layer.file_name()); - let path = match conf.remote_path(&local_path) { - Ok(p) => p, - Err(e) => { - panic!("Can't make a timeline path! {e}"); - } - }; - objects.push(path); + for (layer, generation) in op.layers { + objects.push(remote_layer_path( + &op.tenant_id, + &op.timeline_id, + &layer, + generation, + )); } for path in objects { diff --git a/pageserver/src/deletion_queue/frontend.rs b/pageserver/src/deletion_queue/frontend.rs index edccfc0cc8..0ce47da53e 100644 --- a/pageserver/src/deletion_queue/frontend.rs +++ b/pageserver/src/deletion_queue/frontend.rs @@ -12,12 +12,14 @@ use tokio_util::sync::CancellationToken; use tracing::debug; use tracing::info; use tracing::warn; +use utils::generation::Generation; use utils::id::TenantId; use utils::id::TimelineId; use crate::config::PageServerConf; use crate::metrics::DELETION_QUEUE_ERRORS; use crate::metrics::DELETION_QUEUE_SUBMITTED; +use crate::tenant::remote_timeline_client::remote_layer_path; use crate::tenant::storage_layer::LayerFileName; // The number of keys in a DeletionList before we will proactively persist it @@ -41,7 +43,7 @@ pub(super) struct DeletionOp { // `layers` and `objects` are both just lists of objects. `layers` is used if you do not // have a config object handy to project it to a remote key, and need the consuming worker // to do it for you. - pub(super) layers: Vec, + pub(super) layers: Vec<(LayerFileName, Generation)>, pub(super) objects: Vec, } @@ -306,18 +308,14 @@ impl FrontendQueueWorker { op.objects.len() ); - let timeline_path = self.conf.timeline_path(&op.tenant_id, &op.timeline_id); let mut layer_paths = Vec::new(); - for layer in op.layers { - // TODO go directly to remote path without composing local path - let local_path = timeline_path.join(layer.file_name()); - let path = match self.conf.remote_path(&local_path) { - Ok(p) => p, - Err(e) => { - panic!("Can't make a timeline path! {e}"); - } - }; - layer_paths.push(path); + for (layer, generation) in op.layers { + layer_paths.push(remote_layer_path( + &op.tenant_id, + &op.timeline_id, + &layer, + generation, + )); } self.pending diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs index 4a03faf635..9646262ca4 100644 --- a/pageserver/src/tenant.rs +++ b/pageserver/src/tenant.rs @@ -121,7 +121,7 @@ mod span; pub mod metadata; mod par_fsync; -mod remote_timeline_client; +pub mod remote_timeline_client; pub mod storage_layer; pub mod config; diff --git a/pageserver/src/tenant/remote_timeline_client.rs b/pageserver/src/tenant/remote_timeline_client.rs index 51302066a2..70c25f9e1b 100644 --- a/pageserver/src/tenant/remote_timeline_client.rs +++ b/pageserver/src/tenant/remote_timeline_client.rs @@ -646,7 +646,7 @@ impl RemoteTimelineClient { deletion_queue_client: &DeletionQueueClient, ) -> anyhow::Result<()> { // Synchronous update of upload queues under mutex - let _with_generations = { + let with_generations = { let mut guard = self.upload_queue.lock().unwrap(); let upload_queue = guard.initialized_mut()?; @@ -657,14 +657,14 @@ impl RemoteTimelineClient { // Decorate our list of names with each name's generation, dropping // makes that are unexpectedly missing from our metadata. let with_generations: Vec<_> = names - .iter() + .into_iter() .filter_map(|name| { // Remove from latest_files, learning the file's remote generation in the process let meta = upload_queue.latest_files.remove(name); if let Some(meta) = meta { upload_queue.latest_files_changes_since_metadata_upload_scheduled += 1; - Some((name, meta.generation)) + Some((name.clone(), meta.generation)) } else { // This is unexpected: latest_files is meant to be kept up to // date. We can't delete the layer if we have forgotten what @@ -693,7 +693,7 @@ impl RemoteTimelineClient { // Enqueue deletions deletion_queue_client - .push_layers(self.tenant_id, self.timeline_id, names.to_vec()) + .push_layers(self.tenant_id, self.timeline_id, with_generations) .await?; Ok(()) } @@ -826,7 +826,7 @@ impl RemoteTimelineClient { ) -> anyhow::Result<()> { debug_assert_current_span_has_tenant_and_timeline_id(); - let layers: Vec = { + let layers: Vec<_> = { let mut locked = self.upload_queue.lock().unwrap(); let stopped = locked.stopped_mut()?; @@ -840,20 +840,16 @@ impl RemoteTimelineClient { .upload_queue_for_deletion .latest_files .drain() - .map(|kv| kv.0) + .map(|kv| (kv.0, kv.1.generation)) .collect() }; let layer_deletion_count = layers.len(); - let timeline_path = self.conf.timeline_path(&self.tenant_id, &self.timeline_id); let layer_paths = layers .into_iter() - .map(|l| { - let local_path = timeline_path.join(l.file_name()); - self.conf - .remote_path(&local_path) - .expect("Timeline path should always convert to remote") + .map(|(layer, generation)| { + remote_layer_path(&self.tenant_id, &self.timeline_id, &layer, generation) }) .collect(); deletion_queue.push_immediate(layer_paths).await?; @@ -1711,6 +1707,7 @@ mod tests { "index_part.json", ], &remote_timeline_dir, + generation, ); // Finish uploads and deletions