mirrors/tranquil-pds
1
0
Fork 0
mirror of https://tangled.org/tranquil.farm/tranquil-pds synced 2026-09-21 01:34:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Wrap in counter lock and stop deleting blobs by cid

Browse Source
This commit is contained in:
Johanna Larsson
2026-09-12 15:34:46 +00:00
committed by Tangled
parent 877b587481
commit 2fc5f2e308
8 changed files with 26 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tranquil-db-traits/src/blob.rs
-2
View File
@@ -70,8 +70,6 @@ pub trait BlobRepository: Send + Sync {
takedown_ref: Option<&str>,
) -> Result<bool, DbError>;
async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result<bool, DbError>;
async fn delete_blobs_by_user(&self, user_id: Uuid) -> Result<u64, DbError>;
async fn get_blob_storage_keys_by_user(&self, user_id: Uuid) -> Result<Vec<String>, DbError>;
crates/tranquil-db/src/postgres/blob.rs
-9
View File
@@ -176,15 +176,6 @@ impl BlobRepository for PostgresBlobRepository {
Ok(result.rows_affected() > 0)
}
async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result<bool, DbError> {
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<u64, DbError> {
let result = sqlx::query!("DELETE FROM blobs WHERE created_by_user = $1", user_id)
.execute(&self.pool)
crates/tranquil-pds/tests/store_parity.rs
+1 -9
View File
@@ -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]
crates/tranquil-store/src/metastore/blob_ops.rs
+11 -50
View File
@@ -25,14 +25,21 @@ pub struct BlobOps {
db: Database,
repo_data: Keyspace,
user_hashes: Arc<UserHashMap>,
counter_lock: Arc<parking_lot::Mutex<()>>,
}
impl BlobOps {
pub fn new(db: Database, repo_data: Keyspace, user_hashes: Arc<UserHashMap>) -> Self {
pub fn new(
db: Database,
repo_data: Keyspace,
user_hashes: Arc<UserHashMap>,
counter_lock: Arc<parking_lot::Mutex<()>>,
) -> Self {
Self {
db,
repo_data,
user_hashes,
counter_lock,
}
}
@@ -50,6 +57,7 @@ impl BlobOps {
created_by_user: Uuid,
storage_key: &str,
) -> Result<Option<CidLink>, 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<bool, MetastoreError> {
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<bool, MetastoreError> {
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<u64, MetastoreError> {
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();
crates/tranquil-store/src/metastore/client.rs
-10
View File
@@ -975,16 +975,6 @@ impl<S: StorageIO + 'static> tranquil_db_traits::BlobRepository for MetastoreCli
recv(rx).await
}
async fn delete_blob_by_cid(&self, cid: &CidLink) -> Result<bool, DbError> {
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<u64, DbError> {
let (tx, rx) = oneshot::channel();
self.pool
crates/tranquil-store/src/metastore/handler.rs
+3 -15
View File
@@ -601,10 +601,6 @@ pub enum BlobRequest {
takedown_ref: Option<String>,
tx: Tx<bool>,
},
DeleteBlobByCid {
cid: CidLink,
tx: Tx<bool>,
},
DeleteBlobsByUser {
user_id: Uuid,
tx: Tx<u64>,
@@ -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<S: StorageIO + 'static>(state: &HandlerState<S>, 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
crates/tranquil-store/src/metastore/infra_ops.rs
+10 -47
View File
@@ -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<Option<String>, 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<BlobMetaValue> = point_lookup(
let val: Option<BlobContentValue> = 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(
crates/tranquil-store/src/metastore/mod.rs
+1
View File
@@ -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),
)
}