diff --git a/crates/tranquil-db-traits/src/blob.rs b/crates/tranquil-db-traits/src/blob.rs index 82d0c42..b40e551 100644 --- a/crates/tranquil-db-traits/src/blob.rs +++ b/crates/tranquil-db-traits/src/blob.rs @@ -70,8 +70,6 @@ pub trait BlobRepository: Send + Sync { takedown_ref: Option<&str>, ) -> Result; - async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result; - async fn delete_blobs_by_user(&self, user_id: Uuid) -> Result; async fn get_blob_storage_keys_by_user(&self, user_id: Uuid) -> Result, DbError>; diff --git a/crates/tranquil-db/src/postgres/blob.rs b/crates/tranquil-db/src/postgres/blob.rs index 527885a..546acfe 100644 --- a/crates/tranquil-db/src/postgres/blob.rs +++ b/crates/tranquil-db/src/postgres/blob.rs @@ -176,15 +176,6 @@ impl BlobRepository for PostgresBlobRepository { Ok(result.rows_affected() > 0) } - async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result { - let result = sqlx::query!("DELETE FROM blobs WHERE cid = $1", cid.as_str()) - .execute(&self.pool) - .await - .map_err(map_sqlx_error)?; - - Ok(result.rows_affected() > 0) - } - async fn delete_blobs_by_user(&self, user_id: Uuid) -> Result { let result = sqlx::query!("DELETE FROM blobs WHERE created_by_user = $1", user_id) .execute(&self.pool) diff --git a/crates/tranquil-pds/tests/store_parity.rs b/crates/tranquil-pds/tests/store_parity.rs index 870d023..b696960 100644 --- a/crates/tranquil-pds/tests/store_parity.rs +++ b/crates/tranquil-pds/tests/store_parity.rs @@ -1644,7 +1644,7 @@ async fn parity_plc_tokens() { } #[tokio::test] -async fn parity_blob_delete_and_takedown() { +async fn parity_blob_takedown() { let f = ParityFixture::new().await; let did = test_did("blobdel"); let handle = test_handle("blobdel"); @@ -1681,14 +1681,6 @@ async fn parity_blob_delete_and_takedown() { pg_with_td.as_ref().map(|b| b.takedown_ref.as_deref()), store_with_td.as_ref().map(|b| b.takedown_ref.as_deref()) ); - - f.pg.blob.delete_blob_by_cid(&cid).await.unwrap(); - f.store.blob.delete_blob_by_cid(&cid).await.unwrap(); - - let pg_meta = f.pg.blob.get_blob_metadata(&cid).await.unwrap(); - let store_meta = f.store.blob.get_blob_metadata(&cid).await.unwrap(); - assert!(pg_meta.is_none()); - assert!(store_meta.is_none()); } #[tokio::test] diff --git a/crates/tranquil-store/src/metastore/blob_ops.rs b/crates/tranquil-store/src/metastore/blob_ops.rs index 7a2e5f5..c1a67b5 100644 --- a/crates/tranquil-store/src/metastore/blob_ops.rs +++ b/crates/tranquil-store/src/metastore/blob_ops.rs @@ -25,14 +25,21 @@ pub struct BlobOps { db: Database, repo_data: Keyspace, user_hashes: Arc, + counter_lock: Arc>, } impl BlobOps { - pub fn new(db: Database, repo_data: Keyspace, user_hashes: Arc) -> Self { + pub fn new( + db: Database, + repo_data: Keyspace, + user_hashes: Arc, + counter_lock: Arc>, + ) -> Self { Self { db, repo_data, user_hashes, + counter_lock, } } @@ -50,6 +57,7 @@ impl BlobOps { created_by_user: Uuid, storage_key: &str, ) -> Result, MetastoreError> { + let _guard = self.counter_lock.lock(); if size_bytes < 0 { return Err(MetastoreError::InvalidInput( "size_bytes must be non-negative", @@ -198,6 +206,7 @@ impl BlobOps { cid: &CidLink, takedown_ref: Option<&str>, ) -> Result { + let _guard = self.counter_lock.lock(); let mut content = match self.get_blob_content(cid)? { Some(c) => c, None => return Ok(false), @@ -214,25 +223,8 @@ impl BlobOps { Ok(true) } - pub fn delete_blob_by_cid(&self, cid: &CidLink) -> Result { - let cid_index_key = blob_by_cid_key(cid.as_str()); - if self - .repo_data - .get(cid_index_key.as_slice()) - .map_err(MetastoreError::Fjall)? - .is_none() - { - return Ok(false); - } - - let mut batch = self.db.batch(); - batch.remove(&self.repo_data, cid_index_key.as_slice()); - batch.commit().map_err(MetastoreError::Fjall)?; - - Ok(true) - } - pub fn delete_blobs_by_user(&self, user_id: Uuid) -> Result { + let _guard = self.counter_lock.lock(); let user_hash = self.resolve_user_hash(user_id)?; let prefix = blob_user_prefix(user_hash); @@ -642,37 +634,6 @@ mod tests { assert_eq!(ops.sum_blob_storage().unwrap(), 350); } - #[test] - fn delete_blob_by_cid() { - let (_dir, ms) = open_fresh(); - let (user_id, _) = setup_user(&ms); - let ops = ms.blob_ops(); - - let cid = test_cid_link(40); - ops.insert_blob(&cid, "image/png", 100, user_id, "k") - .unwrap(); - - assert!(ops.delete_blob_by_cid(&cid).unwrap()); - assert!(ops.get_blob_metadata(&cid).unwrap().is_none()); - assert!(!ops.delete_blob_by_cid(&cid).unwrap()); - } - - #[test] - fn delete_blob_cleans_up_indexes() { - let (_dir, ms) = open_fresh(); - let (user_id, _) = setup_user(&ms); - let ops = ms.blob_ops(); - - let cid = test_cid_link(41); - ops.insert_blob(&cid, "image/png", 100, user_id, "storage/abc") - .unwrap(); - assert!(ops.get_blob_storage_key(&cid).unwrap().is_some()); - - ops.delete_blob_by_cid(&cid).unwrap(); - - assert!(ops.get_blob_metadata(&cid).unwrap().is_none()); - } - #[test] fn delete_blobs_by_user() { let (_dir, ms) = open_fresh(); diff --git a/crates/tranquil-store/src/metastore/client.rs b/crates/tranquil-store/src/metastore/client.rs index 2090333..8cfba8a 100644 --- a/crates/tranquil-store/src/metastore/client.rs +++ b/crates/tranquil-store/src/metastore/client.rs @@ -975,16 +975,6 @@ impl tranquil_db_traits::BlobRepository for MetastoreCli recv(rx).await } - async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result { - let (tx, rx) = oneshot::channel(); - self.pool - .send(MetastoreRequest::Blob(BlobRequest::DeleteBlobByCid { - cid: cid.clone(), - tx, - }))?; - recv(rx).await - } - async fn delete_blobs_by_user(&self, user_id: Uuid) -> Result { let (tx, rx) = oneshot::channel(); self.pool diff --git a/crates/tranquil-store/src/metastore/handler.rs b/crates/tranquil-store/src/metastore/handler.rs index 2c76c40..5fe8d87 100644 --- a/crates/tranquil-store/src/metastore/handler.rs +++ b/crates/tranquil-store/src/metastore/handler.rs @@ -601,10 +601,6 @@ pub enum BlobRequest { takedown_ref: Option, tx: Tx, }, - DeleteBlobByCid { - cid: CidLink, - tx: Tx, - }, DeleteBlobsByUser { user_id: Uuid, tx: Tx, @@ -632,9 +628,9 @@ pub enum BlobRequest { impl BlobRequest { fn routing(&self, user_hashes: &UserHashMap) -> Routing { match self { - Self::InsertBlob { cid, .. } - | Self::UpdateBlobTakedown { cid, .. } - | Self::DeleteBlobByCid { cid, .. } => cid_to_routing(cid), + Self::InsertBlob { cid, .. } | Self::UpdateBlobTakedown { cid, .. } => { + cid_to_routing(cid) + } Self::DeleteBlobsByUser { user_id, .. } => uuid_to_routing(user_hashes, user_id), @@ -3267,14 +3263,6 @@ fn dispatch_blob(state: &HandlerState, req: BlobReque .map_err(metastore_to_db); let _ = tx.send(result); } - BlobRequest::DeleteBlobByCid { cid, tx } => { - let result = state - .metastore - .blob_ops() - .delete_blob_by_cid(&cid) - .map_err(metastore_to_db); - let _ = tx.send(result); - } BlobRequest::DeleteBlobsByUser { user_id, tx } => { let result = state .metastore diff --git a/crates/tranquil-store/src/metastore/infra_ops.rs b/crates/tranquil-store/src/metastore/infra_ops.rs index d9f199c..1298b28 100644 --- a/crates/tranquil-store/src/metastore/infra_ops.rs +++ b/crates/tranquil-store/src/metastore/infra_ops.rs @@ -6,7 +6,7 @@ use smallvec::SmallVec; use uuid::Uuid; use super::MetastoreError; -use super::blobs::{BlobMetaValue, blob_by_cid_key, blob_meta_key}; +use super::blobs::{BlobContentValue, blob_by_cid_key}; use super::infra_schema::{ DeletionRequestValue, InviteCodeUseValue, InviteCodeValue, NotificationHistoryValue, QueuedCommsValue, ReportValue, SigningKeyValue, account_pref_key, account_pref_prefix, @@ -1220,57 +1220,20 @@ impl InfraOps { &self, cid: &CidLink, ) -> Result, MetastoreError> { - let cid_str = cid.as_str(); - let cid_index_key = blob_by_cid_key(cid_str); - let user_hash_raw = match self - .repo_data - .get(cid_index_key.as_slice()) - .map_err(MetastoreError::Fjall)? - { - Some(raw) => { - let arr: [u8; 8] = raw - .as_ref() - .try_into() - .map_err(|_| MetastoreError::CorruptData("blob_by_cid value not 8 bytes"))?; - u64::from_be_bytes(arr) - } - None => return Ok(None), - }; - let user_hash = UserHash::from_raw(user_hash_raw); - let key = blob_meta_key(user_hash, cid_str); - let val: Option = point_lookup( + let val: Option = point_lookup( &self.repo_data, - key.as_slice(), - BlobMetaValue::deserialize, - "corrupt blob_meta value", + blob_by_cid_key(cid.as_str()).as_slice(), + BlobContentValue::deserialize, + "corrupt blob_content value", )?; - Ok(val.map(|v| v.storage_key)) + Ok(val.map(|v| v.meta.storage_key)) } pub fn delete_blob_by_cid(&self, cid: &CidLink) -> Result<(), MetastoreError> { - let cid_str = cid.as_str(); - let cid_index_key = blob_by_cid_key(cid_str); - let user_hash_raw = match self - .repo_data - .get(cid_index_key.as_slice()) - .map_err(MetastoreError::Fjall)? - { - Some(raw) => { - let arr: [u8; 8] = raw - .as_ref() - .try_into() - .map_err(|_| MetastoreError::CorruptData("blob_by_cid value not 8 bytes"))?; - u64::from_be_bytes(arr) - } - None => return Ok(()), - }; - let user_hash = UserHash::from_raw(user_hash_raw); - let primary_key = blob_meta_key(user_hash, cid_str); - - let mut batch = self.db.batch(); - batch.remove(&self.repo_data, primary_key.as_slice()); - batch.remove(&self.repo_data, cid_index_key.as_slice()); - batch.commit().map_err(MetastoreError::Fjall) + let _guard = self.counter_lock.lock(); + self.repo_data + .remove(blob_by_cid_key(cid.as_str()).as_slice()) + .map_err(MetastoreError::Fjall) } pub fn get_admin_account_info_by_did( diff --git a/crates/tranquil-store/src/metastore/mod.rs b/crates/tranquil-store/src/metastore/mod.rs index 15f382e..976f587 100644 --- a/crates/tranquil-store/src/metastore/mod.rs +++ b/crates/tranquil-store/src/metastore/mod.rs @@ -354,6 +354,7 @@ impl Metastore { self.db.clone(), self.partitions[Partition::RepoData.index()].clone(), Arc::clone(&self.user_hashes), + Arc::clone(&self.counter_lock), ) }