diff --git a/services/blob/src/database/client.rs b/services/blob/src/database/client.rs index f59fcaf30..13bdc94bf 100644 --- a/services/blob/src/database/client.rs +++ b/services/blob/src/database/client.rs @@ -1,498 +1,534 @@ use aws_sdk_dynamodb::{ operation::put_item::PutItemOutput, types::{ AttributeValue, Delete, DeleteRequest, PutRequest, TransactWriteItem, Update, WriteRequest, }, Error as DynamoDBError, }; use chrono::Utc; -use comm_lib::database::{ - self, batch_operations::ExponentialBackoffConfig, TryFromAttribute, +use comm_lib::{ + blob::types::BlobInfo, + database::{ + self, batch_operations::ExponentialBackoffConfig, AttributeExtractor, + TryFromAttribute, + }, }; use std::collections::HashMap; use tracing::{debug, error, trace}; use crate::constants::db::*; use crate::constants::error_types; use super::errors::{BlobDBError, Error as DBError}; use super::types::*; #[derive(Clone)] pub struct DatabaseClient { ddb: aws_sdk_dynamodb::Client, } /// public interface implementation impl DatabaseClient { pub fn new(aws_config: &aws_config::SdkConfig) -> Self { DatabaseClient { ddb: aws_sdk_dynamodb::Client::new(aws_config), } } /// Gets a blob item row from the database by its blob hash /// Returns None if the blob item is not found pub async fn get_blob_item( &self, blob_hash: impl Into, ) -> DBResult> { let key = PrimaryKey::for_blob_item(blob_hash); self .get_raw_item(key.clone()) .await? .map(BlobItemRow::try_from) .transpose() } /// Inserts a new blob item row into the database. Returns Error /// if the item already exists. pub async fn put_blob_item(&self, blob_item: BlobItemInput) -> DBResult<()> { let item = HashMap::from([ ( ATTR_BLOB_HASH.to_string(), AttributeValue::S(blob_item.blob_hash), ), ( ATTR_HOLDER.to_string(), AttributeValue::S(BLOB_ITEM_ROW_HOLDER_VALUE.into()), ), ( ATTR_S3_PATH.to_string(), AttributeValue::S(blob_item.s3_path.to_full_path()), ), (ATTR_UNCHECKED.to_string(), UncheckedKind::Blob.into()), ]); self.insert_item(item).await?; Ok(()) } /// Deletes blob item row. Doesn't delete its holders. pub async fn delete_blob_item( &self, blob_hash: impl Into, ) -> DBResult<()> { let key = PrimaryKey::for_blob_item(blob_hash); self .ddb .delete_item() .table_name(BLOB_TABLE_NAME) .set_key(Some(key.into())) .send() .await .map_err(|err| { debug!("DynamoDB client failed to delete blob item: {:?}", err); DBError::AwsSdk(Box::new(err.into())) })?; Ok(()) } // Inserts a new holder assignment row into the database. Returns Error // if the item already exists or holder format is invalid. pub async fn put_holder_assignment( &self, blob_hash: impl Into, holder: impl Into, ) -> DBResult<()> { let blob_hash: String = blob_hash.into(); let holder: String = holder.into(); let indexed_tag = get_indexable_tag(&holder, &[]); validate_holder(&holder)?; let mut item = HashMap::from([ (ATTR_BLOB_HASH.to_string(), AttributeValue::S(blob_hash)), (ATTR_HOLDER.to_string(), AttributeValue::S(holder)), (ATTR_UNCHECKED.to_string(), UncheckedKind::Holder.into()), ]); if let Some(tag) = indexed_tag { item.insert(ATTR_INDEXED_TAG.to_string(), AttributeValue::S(tag)); } self.insert_item(item).await?; Ok(()) } /// Deletes a holder assignment row from the table. /// If the blob item for given holder assignment exists, it will be marked as unchecked. /// /// Returns Error if the holder format is invalid or race condition happened. /// Doesn't fail if the holder assignment didn't exist before. pub async fn delete_holder_assignment( &self, blob_hash: impl Into, holder: impl Into, ) -> DBResult<()> { let blob_hash: String = blob_hash.into(); let holder: String = holder.into(); validate_holder(&holder)?; let mut transaction = Vec::new(); // delete the holder row let assignment_key = PrimaryKey { blob_hash: blob_hash.clone(), holder, }; let delete_request = Delete::builder() .table_name(BLOB_TABLE_NAME) .set_key(Some(assignment_key.into())) .build() .expect("key or table_name not set in Delete builder"); transaction .push(TransactWriteItem::builder().delete(delete_request).build()); // mark the blob item as unchecked if exists let blob_primary_key = PrimaryKey::for_blob_item(blob_hash); if self.get_raw_item(blob_primary_key.clone()).await?.is_some() { let update_request = Update::builder() .table_name(BLOB_TABLE_NAME) .set_key(Some(blob_primary_key.into())) // even though we checked that the blob item exists, we still need to check it again // using DDB built-in conditions in case it was deleted in meantime .condition_expression( "attribute_exists(#blob_hash) AND attribute_exists(#holder)", ) .update_expression("SET #unchecked = :unchecked, #last_modified = :now") .expression_attribute_names("#blob_hash", ATTR_BLOB_HASH) .expression_attribute_names("#holder", ATTR_HOLDER) .expression_attribute_names("#unchecked", ATTR_UNCHECKED) .expression_attribute_names("#last_modified", ATTR_LAST_MODIFIED) .expression_attribute_values(":unchecked", UncheckedKind::Blob.into()) .expression_attribute_values( ":now", AttributeValue::N(Utc::now().timestamp_millis().to_string()), ) .build() .expect( "key, table_name or update_expression not set in Update builder", ); transaction .push(TransactWriteItem::builder().update(update_request).build()); } self .ddb .transact_write_items() .set_transact_items(Some(transaction)) .send() .await .map_err(|err| { debug!("DynamoDB client failed to run transaction: {:?}", err); DBError::AwsSdk(Box::new(err.into())) })?; Ok(()) } /// Queries the table for a list of holders for given blob hash. /// Optionally limits the number of results. pub async fn list_blob_holders( &self, blob_hash: impl Into, limit: Option, ) -> DBResult> { let response = self .ddb .query() .table_name(BLOB_TABLE_NAME) .projection_expression("#holder") .key_condition_expression("#blob_hash = :blob_hash") .expression_attribute_names("#blob_hash", ATTR_BLOB_HASH) .expression_attribute_names("#holder", ATTR_HOLDER) .expression_attribute_values( ":blob_hash", AttributeValue::S(blob_hash.into()), ) .consistent_read(true) // we need to increase limit by 1 because the blob item itself can be fetched too // it is filtered-out later .set_limit(limit.map(|it| it + 1)) .send() .await .map_err(|err| { error!( errorType = error_types::DDB_ERROR, "DynamoDB client failed to query holders: {:?}", err ); DBError::AwsSdk(Box::new(err.into())) })?; let Some(items) = response.items else { return Ok(vec![]); }; items .into_iter() .filter_map(|mut row| { // filter out rows that are blob items // we cannot do it in key condition expression - it doesn't support the <> operator // filter expression doesn't work either - it doesn't support filtering by sort key match String::try_from_attr(ATTR_HOLDER, row.remove(ATTR_HOLDER)) { Ok(value) if value.as_str() == BLOB_ITEM_ROW_HOLDER_VALUE => None, holder => Some(holder), } }) .collect::, _>>() .map_err(DBError::Attribute) } /// Returns a list of primary keys for rows that already exist in the table pub async fn list_existing_keys( &self, keys: impl IntoIterator, ) -> DBResult> { database::batch_operations::batch_get( &self.ddb, BLOB_TABLE_NAME, keys, Some(format!("{}, {}", ATTR_BLOB_HASH, ATTR_HOLDER)), ExponentialBackoffConfig::default(), ) .await? .into_iter() .map(PrimaryKey::try_from) .collect::, _>>() } + pub async fn query_indexed_holders( + &self, + tag: String, + ) -> DBResult> { + self + .ddb + .query() + .table_name(BLOB_TABLE_NAME) + .index_name(HOLDER_TAG_INDEX_NAME) + .key_condition_expression("#indexed_tag = :tag") + .expression_attribute_names("#indexed_tag", HOLDER_TAG_INDEX_KEY_ATTR) + .expression_attribute_values(":tag", AttributeValue::S(tag)) + .send() + .await + .map_err(|err| { + error!( + errorType = error_types::DDB_ERROR, + "DynamoDB client failed to query indexed holders: {:?}", err + ); + DBError::AwsSdk(Box::new(err.into())) + })? + .items + .unwrap_or_default() + .into_iter() + .map(|mut item| { + let blob_hash = item.take_attr(ATTR_BLOB_HASH)?; + let holder = item.take_attr(ATTR_HOLDER)?; + Ok(BlobInfo { blob_hash, holder }) + }) + .collect() + } + /// Returns a list of primary keys for "unchecked" items (blob / holder) /// that were last modified at least `min_age` ago. /// We need to specify if we want to get blob or holder items. pub async fn find_unchecked_items( &self, kind: UncheckedKind, min_age: chrono::Duration, ) -> DBResult> { let created_until = Utc::now() - min_age; let timestamp = created_until.timestamp_millis(); let response = self .ddb .query() .table_name(BLOB_TABLE_NAME) .index_name(UNCHECKED_INDEX_NAME) .key_condition_expression( "#unchecked = :kind AND #last_modified < :timestamp", ) .expression_attribute_names("#unchecked", ATTR_UNCHECKED) .expression_attribute_names("#last_modified", ATTR_LAST_MODIFIED) .expression_attribute_values(":kind", kind.into()) .expression_attribute_values( ":timestamp", AttributeValue::N(timestamp.to_string()), ) .send() .await .map_err(|err| { error!( errorType = error_types::DDB_ERROR, "DynamoDB client failed to query unchecked items: {:?}", err ); DBError::AwsSdk(Box::new(err.into())) })?; let Some(items) = response.items else { return Ok(vec![]); }; items .into_iter() .map(PrimaryKey::try_from) .collect::, _>>() } /// For all rows in specified set of primary keys, removes /// the "unchecked" attribute using PutItem operation in batch. pub async fn batch_mark_checked( &self, keys: impl IntoIterator, ) -> DBResult<()> { let items_to_mark = database::batch_operations::batch_get( &self.ddb, BLOB_TABLE_NAME, keys, None, ExponentialBackoffConfig::default(), ) .await?; let write_requests = items_to_mark .into_iter() .filter_map(|mut row| { // filter out rows that are already checked // to save some write capacity row.remove(ATTR_UNCHECKED)?; let put_request = PutRequest::builder() .set_item(Some(row)) .build() .expect("item not set in PutRequest builder"); let request = WriteRequest::builder().put_request(put_request).build(); Some(request) }) .collect(); database::batch_operations::batch_write( &self.ddb, BLOB_TABLE_NAME, write_requests, ExponentialBackoffConfig::default(), ) .await?; Ok(()) } /// Performs multiple DeleteItem operations in batch pub async fn batch_delete_rows( &self, keys: impl IntoIterator, ) -> DBResult<()> { let write_requests = keys .into_iter() .map(|key| { DeleteRequest::builder() .set_key(Some(key.into())) .build() .expect("key not set in DeleteRequest builder") }) .map(|request| WriteRequest::builder().delete_request(request).build()) .collect::>(); database::batch_operations::batch_write( &self.ddb, BLOB_TABLE_NAME, write_requests, ExponentialBackoffConfig::default(), ) .await?; Ok(()) } } // private helpers impl DatabaseClient { /// inserts a new item into the table using PutItem. Returns /// error if the item already exists async fn insert_item( &self, mut item: RawAttributes, ) -> DBResult { // add metadata attributes common for all types of rows let now = Utc::now().timestamp_millis(); item.insert( ATTR_CREATED_AT.to_string(), AttributeValue::N(now.to_string()), ); item.insert( ATTR_LAST_MODIFIED.to_string(), AttributeValue::N(now.to_string()), ); self .ddb .put_item() .table_name(BLOB_TABLE_NAME) .set_item(Some(item)) // make sure we don't accidentaly overwrite existing row .condition_expression( "attribute_not_exists(#blob_hash) AND attribute_not_exists(#holder)", ) .expression_attribute_names("#blob_hash", ATTR_BLOB_HASH) .expression_attribute_names("#holder", ATTR_HOLDER) .send() .await .map_err(|err| match DynamoDBError::from(err) { DynamoDBError::ConditionalCheckFailedException(e) => { debug!("DynamoDB client failed to insert: item already exists"); trace!("Conditional check failed with error: {}", e); DBError::ItemAlreadyExists } err => { debug!("DynamoDB client failed to insert: {:?}", err); DBError::AwsSdk(Box::new(err)) } }) } /// Gets a single row from the table using GetItem, without parsing it async fn get_raw_item( &self, key: PrimaryKey, ) -> DBResult> { self .ddb .get_item() .table_name(BLOB_TABLE_NAME) .set_key(Some(key.into())) .send() .await .map_err(|err| { debug!("DynamoDB client failed to get item: {:?}", err); DBError::AwsSdk(Box::new(err.into())) }) .map(|response| response.item) } } fn validate_holder(holder: &str) -> DBResult<()> { if holder == BLOB_ITEM_ROW_HOLDER_VALUE { debug!("Invalid holder: {}", holder); return Err(DBError::Blob(BlobDBError::InvalidInput(holder.to_string()))); } Ok(()) } /// In the future we'll want to add a `tags` meta attribute for internal /// use, e.g. in case of data loss on other services. The attribute /// is going to be a list of string values - _tags_. /// First tag from the list is going to be indexed in DDB. /// /// While tags are not implemented, we accept _holder prefixes_ /// as an intermediate solution - holder prefix (separated by the `:` character) /// is treated as the first holder _tag_, unless [`tags`] is not empty. fn get_indexable_tag(holder: &str, tags: &[String]) -> Option { const HOLDER_PREFIX_SEPARATOR: char = ':'; if let Some(first_tag) = tags.first() { return Some(first_tag.to_string()); } if !holder.contains(HOLDER_PREFIX_SEPARATOR) { return None; } holder .split(HOLDER_PREFIX_SEPARATOR) .next() .map(str::to_string) } #[cfg(test)] mod tests { use super::*; #[test] fn get_indexable_tag_no_tags_no_prefix() { let tag = get_indexable_tag("foo", &[]); assert!(tag.is_none()); } #[test] fn get_indexable_tag_tags_no_prefix() { let tags = vec!["tag1".to_string(), "tag2".to_string()]; let tag = get_indexable_tag("foo", &tags); assert_eq!(tag, Some("tag1".into())); } #[test] fn get_indexable_tag_tags_and_prefix() { let tags = vec!["tag1".to_string(), "tag2".to_string()]; let tag = get_indexable_tag("device1:foo", &tags); assert_eq!(tag, Some("tag1".into())); } #[test] fn get_indexable_tag_prefix_no_tags() { let tag = get_indexable_tag("device1:foo", &[]); assert_eq!(tag, Some("device1".into())); } } diff --git a/services/blob/src/service.rs b/services/blob/src/service.rs index 86f1d73de..0b5524272 100644 --- a/services/blob/src/service.rs +++ b/services/blob/src/service.rs @@ -1,659 +1,668 @@ #![allow(unused)] +use comm_lib::blob::types::BlobInfo; use regex::RegexSet; use std::collections::{BTreeMap, HashMap, HashSet}; use std::ops::{Bound, Range, RangeBounds, RangeInclusive}; use std::sync::Arc; use async_stream::try_stream; use chrono::Duration; use comm_lib::http::ByteStream; use comm_lib::shared::reserved_users::RESERVED_USERNAME_SET; use comm_lib::tools::BoxedError; use once_cell::sync::Lazy; use tokio_stream::StreamExt; use tonic::codegen::futures_core::Stream; use tracing::{debug, error, info, trace, warn}; use crate::config::{CONFIG, OFFENSIVE_INVITE_LINKS}; use crate::constants::{ INVITE_LINK_BLOB_HASH_PREFIX, S3_MULTIPART_UPLOAD_MINIMUM_CHUNK_SIZE, }; use crate::database::types::{ BlobItemInput, BlobItemRow, PrimaryKey, UncheckedKind, }; use crate::database::DBError; use crate::s3::{Error as S3Error, S3Client, S3Path}; use crate::tools::MemOps; use crate::{ constants::error_types, constants::BLOB_DOWNLOAD_CHUNK_SIZE, database::DatabaseClient, }; #[derive( Debug, derive_more::Display, derive_more::From, derive_more::Error, )] pub enum InviteLinkError { Reserved, Offensive, } #[derive( Debug, derive_more::Display, derive_more::From, derive_more::Error, )] pub enum BlobServiceError { BlobNotFound, BlobAlreadyExists, InvalidState, DB(DBError), S3(S3Error), InputError(#[error(ignore)] BoxedError), InviteLinkError(InviteLinkError), } type BlobServiceResult = Result; #[derive(Clone, Debug)] pub struct BlobServiceConfig { /// Blob data is streamed from S3 in chunks of this size. pub download_chunk_size: usize, /// If enabled, orphaned blobs will be deleted immediately after /// last holder is removed. This option should be enabled /// if maintenance garbage collection tasks are not run. pub instant_delete_orphaned_blobs: bool, /// Minimum age that a orphan must stay unmodified /// before it can be deleted by a garbage collection task /// This option is ignored if `instant_delete_orphaned_blobs` is `true` pub orphan_protection_period: chrono::Duration, } static OFFENSIVE_INVITE_LINKS_REGEX_SET: Lazy = Lazy::new(|| { RegexSet::new(OFFENSIVE_INVITE_LINKS.iter().collect::>()).unwrap() }); impl Default for BlobServiceConfig { fn default() -> Self { BlobServiceConfig { download_chunk_size: BLOB_DOWNLOAD_CHUNK_SIZE as usize, instant_delete_orphaned_blobs: false, orphan_protection_period: Duration::hours(1), } } } #[derive(Clone)] pub struct BlobService { db: Arc, s3: S3Client, config: BlobServiceConfig, } impl BlobService { pub fn new( db: DatabaseClient, s3: S3Client, config: BlobServiceConfig, ) -> Self { Self { db: Arc::new(db), s3, config, } } /// Retrieves blob object metadata and returns a download object /// that can be used to download the blob data. pub async fn create_download( &self, blob_hash: impl Into, ) -> BlobServiceResult { // 1. Get S3 path let s3_path = match self.db.get_blob_item(blob_hash.into()).await { Ok(Some(BlobItemRow { s3_path, .. })) => Ok(s3_path), Ok(None) => { debug!("Blob not found"); Err(BlobServiceError::BlobNotFound) } Err(err) => Err(BlobServiceError::DB(err)), }?; debug!("S3 path: {:?}", s3_path); // 2. Get S3 Object metadata trace!("Getting S3 object metadata..."); let object_metadata = self.s3.get_object_metadata(&s3_path).await?; let blob_size = object_metadata .content_length() .ok_or_else(|| { error!( errorType = error_types::S3_ERROR, "Failed to get S3 object content length" ); BlobServiceError::InvalidState }) .and_then(|len| { if len >= 0 { Ok(len as u64) } else { error!( errorType = error_types::S3_ERROR, "S3 object content length is negative" ); Err(BlobServiceError::InvalidState) } })?; debug!("S3 object size: {} bytes", blob_size); // 3. Create download session let session = BlobDownloadObject { s3_path, blob_size, byte_range: 0..blob_size, chunk_size: self.config.download_chunk_size as u64, s3_client: self.s3.clone(), }; Ok(session) } fn validate_invite_link_blob_hash( invite_secret: &str, ) -> Result<(), BlobServiceError> { let lowercase_secret = invite_secret.to_lowercase(); if (OFFENSIVE_INVITE_LINKS_REGEX_SET.is_match(&lowercase_secret)) { debug!("Offensive invite link"); return Err(BlobServiceError::InviteLinkError( InviteLinkError::Offensive, )); } Ok(()) } pub async fn put_blob( &self, blob_hash: impl Into, mut blob_data_stream: impl ByteStream, ) -> Result<(), BlobServiceError> { let blob_hash: String = blob_hash.into(); let blob_item = BlobItemInput::new(&blob_hash); if self.db.get_blob_item(&blob_hash).await?.is_some() { debug!("Blob already exists"); return Err(BlobServiceError::BlobAlreadyExists); } if let Some(invite_secret) = blob_hash.strip_prefix(INVITE_LINK_BLOB_HASH_PREFIX) { Self::validate_invite_link_blob_hash(invite_secret)?; } let mut upload_session = self.s3.start_upload_session(&blob_item.s3_path).await?; trace!(?blob_item, "Started S3 upload session"); tokio::pin!(blob_data_stream); let mut s3_chunk: Vec = Vec::new(); while let Some(chunk) = blob_data_stream.try_next().await.map_err(|err| { warn!("Failed to get data chunk: {:?}", err); BlobServiceError::InputError(err) })? { s3_chunk.extend_from_slice(&chunk); // New parts should be added to AWS only if they exceed minimum part size, // Otherwise AWS returns error if s3_chunk.len() as u64 > S3_MULTIPART_UPLOAD_MINIMUM_CHUNK_SIZE { trace!( chunk_size = s3_chunk.len(), "Chunk size exceeded, adding new S3 part" ); upload_session .add_part(s3_chunk.take_out()) .await .map_err(BlobServiceError::from)?; } } trace!("Upload stream drained"); // add the remaining data as the last S3 part if !s3_chunk.is_empty() { trace!("Uploading remaining {} bytes", s3_chunk.len()); upload_session.add_part(s3_chunk).await?; } // Complete the upload session upload_session.finish_upload().await?; trace!("S3 upload complete, putting item to db"); self.db.put_blob_item(blob_item).await?; Ok(()) } pub async fn assign_holder( &self, blob_hash: impl Into, holder: impl Into, ) -> BlobServiceResult<()> { let blob_hash: String = blob_hash.into(); trace!(blob_hash, "Attempting to assign holder"); self .db .put_holder_assignment(blob_hash.clone(), holder.into()) .await?; trace!("Holder assigned."); Ok(()) } pub async fn revoke_holder( &self, blob_hash: impl Into, holder: impl Into, instant_delete: bool, ) -> BlobServiceResult<()> { let blob_hash: String = blob_hash.into(); let holder: String = holder.into(); trace!(blob_hash, holder, "Attempting to revoke holder"); self.db.delete_holder_assignment(&blob_hash, holder).await?; if instant_delete || self.config.instant_delete_orphaned_blobs { trace!("Instant orphan deletion enabled. Looking for holders"); let is_orphan = self .db .list_blob_holders(&blob_hash, Some(1)) .await? .is_empty(); if !is_orphan { trace!("Found holders, nothing to do"); return Ok(()); } debug!("No holders left, deleting blob if exists"); trace!("Getting blob item"); let Some(blob_item) = self.db.get_blob_item(&blob_hash).await? else { trace!("Blob item not found, nothing to do"); return Ok(()); }; trace!("Deleting S3 object"); self.s3.delete_object(&blob_item.s3_path).await?; trace!("Deleting blob item entry from DB"); self.db.delete_blob_item(blob_hash).await?; } Ok(()) } pub async fn find_existing_blobs( &self, blob_hashes: HashSet<&String>, ) -> BlobServiceResult> { let primary_keys = blob_hashes.into_iter().map(PrimaryKey::for_blob_item); let existing_items: HashSet = self .db .list_existing_keys(primary_keys) .await? .into_iter() .map(|pk| pk.blob_hash) .collect(); Ok(existing_items) } + pub async fn query_indexed_holders( + &self, + tag: String, + ) -> BlobServiceResult> { + let results = self.db.query_indexed_holders(tag).await?; + Ok(results) + } + pub async fn perform_cleanup(&self) -> anyhow::Result<()> { info!("Starting cleanup..."); // 1. Fetch blobs and holders marked as "unchecked" debug!("Querying for unchecked blobs and holders..."); let protection_periond = self.config.orphan_protection_period; let (unchecked_blobs, unchecked_holders) = tokio::try_join!( self .db .find_unchecked_items(UncheckedKind::Blob, protection_periond), self .db .find_unchecked_items(UncheckedKind::Holder, protection_periond) )?; debug!( "Found {} unchecked blobs and {} unchecked holders", unchecked_blobs.len(), unchecked_holders.len() ); let mut unchecked_items = UncheckedCollection::new(); // 2. construct structures of possibly orphaned blobs debug!("Creating structures of possibly orphaned items..."); for PrimaryKey { blob_hash, .. } in unchecked_blobs { trace!("Creating unchecked item for blob hash '{}'", &blob_hash); unchecked_items.insert( blob_hash.clone(), UncheckedItem { blob_hash: Some(blob_hash), holders: Vec::new(), }, ); } // 3. iterate over possibly orphaned holders and fill the structs for PrimaryKey { blob_hash, holder } in unchecked_holders { if let Some(item) = unchecked_items.get_mut(&blob_hash) { trace!( "Inserting holder '{}' for blob hash '{}'", &holder, &blob_hash ); item.holders.push(holder); } else { trace!( "Creating empty item for holder '{}' (blob hash: '{}')", &holder, &blob_hash ); unchecked_items.insert( blob_hash.clone(), UncheckedItem { blob_hash: None, holders: vec![holder], }, ); } } let mut orphans = HashSet::new(); let mut checked = HashSet::new(); // 4. Filter out items that are for sure not orphaned let checked_items = unchecked_items.filter_out_checked(); debug!("Filtered out {} checked items", checked_items.len()); checked.extend(checked_items); // 5. Query DDB for additional blobs and holders to check if they exist let mut fetch_results = Vec::new(); // 5a. Query holders - Find if possibly orphan blobs have at least one holder debug!("Querying holders for possibly orphaned blobs..."); for blob_hash in unchecked_items.blobs_to_find_holders() { let holders = self .db .list_blob_holders(blob_hash, Some(1)) .await? .into_iter() .map(|holder| PrimaryKey::new(blob_hash.to_string(), holder)); let len_before = fetch_results.len(); fetch_results.extend(holders); trace!( "Found {} holders for blob hash '{}'", fetch_results.len() - len_before, blob_hash ); } // 5b. Query blobs - Find if possibly orphaned holders have blobs debug!("Querying blobs for possibly orphaned holders..."); let blobs_to_get = unchecked_items.blobs_to_check_existence(); let queried_blobs_len = blobs_to_get.len(); let existing_blobs = self.db.list_existing_keys(blobs_to_get).await?; debug!( "Found {} existing blobs out of {} queried", existing_blobs.len(), queried_blobs_len ); fetch_results.extend(existing_blobs); // 6. Update the struct with query results // Then do 2nd pass of filtering out checked items (repeat step 4) debug!("Feeding data structure with query results and filtering again..."); unchecked_items.feed_with_query_results(fetch_results); let checked_items = unchecked_items.filter_out_checked(); debug!("Filtered out {} checked items", checked_items.len()); checked.extend(checked_items); // 7. Perform actual cleanup orphans.extend(unchecked_items.into_primary_keys()); let s3_paths: Vec = orphans .iter() .filter(|pk| pk.is_blob_item()) .map(|PrimaryKey { blob_hash, .. }| S3Path { bucket_name: CONFIG.s3_bucket_name.clone(), object_name: blob_hash.to_string(), }) .collect(); let num_orphans = orphans.len(); let num_checked = checked.len(); let num_s3_blobs = s3_paths.len(); // 7a. Make changes to database debug!("Cleaning up database... Marking {} items as checked and deleting {} orphans", num_checked, num_orphans); tokio::try_join!( self.db.batch_delete_rows(orphans), self.db.batch_mark_checked(checked) )?; // 7b. Delete orphaned blobs from S3 debug!("Cleaning up S3... Deleting {} blobs", num_s3_blobs); self.s3.batch_delete_objects(s3_paths).await?; info!( "Cleanup complete. Deleted orphaned {} DB items and marked {} items as checked. {} blobs were deleted from S3", num_orphans, num_checked, num_s3_blobs ); Ok(()) } } // A B-tree map performs well for both random and sequential access. type BlobHash = String; type UncheckedCollection = BTreeMap; /// Represents an "unchecked" blob entity. It might miss either /// blob hash or holders. #[derive(Debug)] struct UncheckedItem { blob_hash: Option, holders: Vec, } impl UncheckedItem { fn has_blob_hash(&self) -> bool { self.blob_hash.is_some() } fn has_holders(&self) -> bool { !self.holders.is_empty() } /// Returns primary keys for this item. It contains primary heys for holders /// and for blob item (if it has hash). /// A fallback hash is required for holders if item's blob hash is None. fn as_primary_keys(&self, fallback_blob_hash: &str) -> Vec { if !self.has_holders() && !self.has_blob_hash() { warn!( fallback_blob_hash, "Item has no hash and no holders, this should never happen!" ); return Vec::new(); } let hash_for_holders = self.blob_hash.as_deref().unwrap_or(fallback_blob_hash); let mut keys = self .holders .iter() .map(|holder| PrimaryKey { blob_hash: hash_for_holders.to_string(), holder: holder.to_string(), }) .collect::>(); if let Some(blob_hash) = &self.blob_hash { keys.push(PrimaryKey::for_blob_item(blob_hash.to_string())); } keys } } trait CleanupOperations { /// Retains only items that should remain unchecked /// (missing blob hash or holders). /// /// Returns removed items - these items are checked /// (contain both blob hash and at least one holder). fn filter_out_checked(&mut self) -> Vec; /// Returns list of blob hashes for which we need to query if they contain /// at least one holder fn blobs_to_find_holders(&self) -> Vec<&BlobHash>; /// Returns primary keys for blob items that need to be checked if they exist /// /// Technically, this returns all items that have holders but no hash. fn blobs_to_check_existence(&self) -> Vec; /// Updates the structure after fetching additional data from database. fn feed_with_query_results( &mut self, fetched_items: impl IntoIterator, ); /// Turns this collection into a list of DB primary keys fn into_primary_keys(self) -> Vec; } impl CleanupOperations for UncheckedCollection { /// Retains only items that should remain unchecked /// (missing blob hash or holders). /// /// Returns removed items - these items are checked /// (contain both blob hash and at least one holder). fn filter_out_checked(&mut self) -> Vec { let mut checked = Vec::new(); self.retain(|blob_hash, item| { if !item.has_blob_hash() || !item.has_holders() { // blob hash or holder missing, leave unchecked return true; } checked.extend(item.as_primary_keys(blob_hash)); false }); checked } /// Returns list of blob hashes for which we need to query if they contain /// at least one holder fn blobs_to_find_holders(&self) -> Vec<&BlobHash> { self .iter() .filter_map(|(blob_hash, item)| { if item.has_blob_hash() && !item.has_holders() { Some(blob_hash) } else { None } }) .collect() } /// Returns primary keys for blob items that need to be checked if they exist /// /// Technically, this returns all blob items that have holders but no hash. fn blobs_to_check_existence(&self) -> Vec { self .iter() .filter_map(|(blob_hash, item)| { if item.has_holders() && !item.has_blob_hash() { Some(PrimaryKey::for_blob_item(blob_hash)) } else { None } }) .collect() } /// Updates the structure after fetching additional data from database. fn feed_with_query_results( &mut self, fetched_items: impl IntoIterator, ) { for pk in fetched_items.into_iter() { let Some(item) = self.get_mut(&pk.blob_hash) else { warn!("Got fetched item that was not requested: {:?}", pk); continue; }; if pk.is_blob_item() { item.blob_hash = Some(pk.blob_hash) } else { item.holders.push(pk.holder); } } } fn into_primary_keys(self) -> Vec { self .into_iter() .flat_map(|(blob_hash, item)| item.as_primary_keys(&blob_hash)) .collect() } } pub struct BlobDownloadObject { /// Size of the whole blob object in bytes. pub blob_size: u64, /// Range of bytes to be downloaded (exclusive end). byte_range: Range, chunk_size: u64, s3_client: S3Client, s3_path: S3Path, } impl BlobDownloadObject { pub fn set_byte_range(&mut self, range: impl RangeBounds) { let range_start = match range.start_bound() { Bound::Included(&start) => start, Bound::Excluded(&start) => start + 1, Bound::Unbounded => 0, }; let range_end = match range.end_bound() { Bound::Included(&end) => end + 1, Bound::Excluded(&end) => end, Bound::Unbounded => self.blob_size, }; // Clamp range to blob size let start = std::cmp::max(range_start, 0); let end_exclusive = std::cmp::min(range_end, self.blob_size); self.byte_range = start..end_exclusive; debug!("Requested byte range: {}..{}", start, end_exclusive); } /// Size of the data to be downloaded in bytes. pub fn download_size(&self) -> u64 { self.byte_range.end - self.byte_range.start } pub fn into_stream(self) -> impl Stream>> { let BlobDownloadObject { byte_range, chunk_size, s3_path, s3_client, .. } = self; try_stream! { trace!("Starting download stream"); let mut offset: u64 = byte_range.start; while offset < byte_range.end { let next_size = std::cmp::min(chunk_size, byte_range.end - offset); let range = offset..(offset + next_size); trace!(?range, "Getting {} bytes of data", next_size); yield s3_client.get_object_bytes(&s3_path, range).await?; offset += next_size; } } } }