From 99dc5a5c279d975c06061e4a712a0f41705a8b48 Mon Sep 17 00:00:00 2001 From: John Spray Date: Wed, 16 Aug 2023 10:26:33 +0100 Subject: [PATCH] Deletion queue: implement recovery on startup --- pageserver/src/config.rs | 12 +++- pageserver/src/deletion_queue.rs | 109 ++++++++++++++++++++++++++++--- 2 files changed, 110 insertions(+), 11 deletions(-) diff --git a/pageserver/src/config.rs b/pageserver/src/config.rs index 05c18b937e..5808a325bb 100644 --- a/pageserver/src/config.rs +++ b/pageserver/src/config.rs @@ -566,11 +566,19 @@ impl PageServerConf { self.workdir.join("tenants") } - pub fn remote_deletion_list_path(&self, sequence: u64) -> PathBuf { + pub fn remote_deletion_node_prefix(&self) -> PathBuf { PathBuf::from_str("deletion") .unwrap() .join(self.id.to_string()) - .join(sequence.to_string()) + } + + pub fn remote_deletion_list_path(&self, sequence: u64) -> RemotePath { + RemotePath::new( + &self + .remote_deletion_node_prefix() + .join(format!("{:016x}.list", sequence)), + ) + .expect("This should always be convertible, it is relative") } pub fn tenant_path(&self, tenant_id: &TenantId) -> PathBuf { diff --git a/pageserver/src/deletion_queue.rs b/pageserver/src/deletion_queue.rs index f35d07cbf7..969b519d22 100644 --- a/pageserver/src/deletion_queue.rs +++ b/pageserver/src/deletion_queue.rs @@ -6,6 +6,7 @@ use serde_with::serde_as; use tokio; use tokio::time::Duration; use tracing::{self, debug, error, info, warn}; +use utils::backoff; use utils::id::{TenantId, TimelineId}; use crate::{config::PageServerConf, tenant::storage_layer::LayerFileName}; @@ -331,10 +332,7 @@ impl BackendQueueWorker { .iter() .rev() .take(MAX_KEYS_PER_DELETE) - .map(|l| { - RemotePath::new(&self.conf.remote_deletion_list_path(l.sequence)) - .expect("Failed to compose deletion list path") - }) + .map(|l| self.conf.remote_deletion_list_path(l.sequence)) .collect(); match self.remote_storage.delete_objects(&executed_keys).await { @@ -398,8 +396,7 @@ impl FrontendQueueWorker { /// This does not return errors, because on failure to flush we do not lose /// any state: flushing will be retried implicitly on the next deadline async fn flush(&mut self) { - let key = RemotePath::new(&self.conf.remote_deletion_list_path(self.pending.sequence)) - .expect("Failed to compose deletion list path"); + let key = &self.conf.remote_deletion_list_path(self.pending.sequence); let bytes = serde_json::to_vec(&self.pending).expect("Failed to serialize deletion list"); let size = bytes.len(); @@ -458,10 +455,106 @@ impl FrontendQueueWorker { } } + async fn recover(&mut self) -> Result<(), anyhow::Error> { + // TODO: this needs a CancellationToken or equivalent: usual worker teardown happens via the channel + let prefix = RemotePath::new(&self.conf.remote_deletion_node_prefix()) + .expect("Failed to compose path"); + let lists = backoff::retry( + || async { self.remote_storage.list_prefixes(Some(&prefix)).await }, + |_| false, // TODO impl is_permanent + 3, + u32::MAX, // There's no point giving up, since once we do that the deletion queue is stuck + "Recovering deletion lists", + ) + .await?; + + const LIST_EXTENSION: &str = "list"; + + let mut seqs: Vec = Vec::new(); + for l in &lists { + let basename = l + .strip_prefix(&prefix) + .expect("Stripping prefix frrom a prefix listobjects should always work"); + let basename = match basename.to_str() { + Some(s) => s, + None => { + // Should never happen, we are the only ones writing objects here + warn!("Unexpected key encoding in deletion queue object"); + continue; + } + }; + + let (seq_part, extension) = match basename.split_once(".") { + Some(parts) => parts, + None => { + warn!("Unexpected key in deletion queue: {basename}"); + continue; + } + }; + + if extension != LIST_EXTENSION { + continue; + } + + let seq: u64 = match seq_part.parse() { + Ok(s) => s, + Err(e) => { + warn!("Malformed key '{basename}': {e}"); + continue; + } + }; + seqs.push(seq); + } + + seqs.sort(); + + for s in seqs { + let list_path = self.conf.remote_deletion_list_path(s); + let lists_body = backoff::retry( + || self.remote_storage.download_all(&list_path), + |_| false, + 3, + u32::MAX, + "Reading a deletion list", + ) + .await?; + + let deletion_list = match serde_json::from_slice::(lists_body.as_slice()) + { + Ok(l) => l, + Err(e) => { + // Drop the list on the floor: any objects it referenced will be left behind + // for scrubbing to clean up. This should never happen unless we have a serialization bug. + warn!(sequence = s, "Failed to deserialize deletion list: {e}"); + continue; + } + }; + + // We will drop out of recovery if this fails: it indicates that we are shutting down + // or the backend has panicked + self.tx + .send(BackendQueueMessage::Delete(deletion_list)) + .await?; + + self.pending.sequence = s + 1; + } + + Ok(()) + } + /// This is the front-end ingest, where we bundle up deletion requests into DeletionList /// and write them out, for later pub async fn background(&mut self) { info!("Started deletion frontend worker"); + + // Before accepting any input from this pageserver lifetime, recover all deletion lists that are in S3 + if let Err(e) = self.recover().await { + // This should only happen in truly unrecoverable cases, like the recovery finding that the backend + // queue receiver has been dropped. + info!("Deletion queue recover aborted, deletion queue will not proceed ({e:#})"); + return; + } + loop { let msg = match tokio::time::timeout(self.timeout, self.rx.recv()).await { Ok(Some(msg)) => msg, @@ -561,9 +654,7 @@ impl DeletionQueue { ( Self { tx }, Some(FrontendQueueWorker { - // TODO: on startup, recover sequence number by listing persistent list objects, - // *or* if we implement generation numbers, we may start from 0 every time - pending: DeletionList::new(0xdeadbeef), + pending: DeletionList::new(1), remote_storage: remote_storage.clone(), conf, rx,