From 695a7d981c70527423c8f56b7a7c7c342db72391 Mon Sep 17 00:00:00 2001 From: Johanna Larsson Date: Tue, 8 Sep 2026 15:37:03 +0100 Subject: [PATCH] Allow blobs to be owned by multiple accounts The current tranquil database design only allows each blob to be owned by one account. This means that if a second account also has that blob, tranquil skips associated the blob with the account. That works fine a lot of the time, since blobs are looked up by cid and the blob exists. However, it can lead to loss of data under certain scenarios. One example is where I upload a blob, the blob already exists in my instance so insertion is skipped (postgres requires cid to be unique in blobs, fjall only allows one owner per blob cid), I then decide to migrate off tranquil, the blob does not come with me since it is not mine. Another example is where an account is deleted. If a blob was uploaded for account a, then account b uploads the same blob tranquil skips storing it since it exists. Then I delete account a, now account b's blob is missing. I accidentally stumbled upon this when I migrated my account to my own tranuil instance and list blobs now lists 2 fewer blobs than before, two images that had been uploaded by accounts already on the PDS. ps I found record_blobs a bit confusing, at first it looked like a blob ownership table, but then it turns out to just be used for migrations! This PR makes the blob primary key be cid+user for postgres, and updates the queries to account for there being multiple "blobs" with the same cid. For queries that just care about the blob existing, it doesn't matter "whose" blob it is, so limit 1. Most of the work is on the metastore side. Adds ref_count to track how many are referencing the blob since we can't just check for other rows. Instead of storing blobs directly, we now store a per account cid, and the blob reference itself is shared and keyed by cid only. This means some of these operations now require updating two places, so they're done in `batch`es. With the new layout get_blob_value becomes simpler, all blob data is a single "table" or whatever it's called, so we just grab it using cid instead of looking it up for the user. Migrates blobs rather than maintaining two different versions of the tables, although it seems like that could be supported. I removed a test that asserted the old behavior, and added a reasonable (?) set of new tests that assert the new behavior, including a parity test. --- ...af1730906d889101b9039113ec8aa234618d.json} | 4 +- ...8a76e92065e6ae4a6081bd38a19821fbfcc3.json} | 4 +- ...75dbf7bf6da8d903145b12871dda6e1fd5d9.json} | 4 +- ...624ee07a43c380d2df31efdba5cf299aab571.json | 22 ++ ...1e377da6a187dfa3347bd285078a7b4410f2.json} | 4 +- ...488ce7e5b886dba0529fb073c3e0461fe030.json} | 4 +- ...06dacd72b33d04c1e2b98475018ad25485852.json | 22 -- crates/tranquil-api/src/repo/blob.rs | 27 +- crates/tranquil-db/src/postgres/blob.rs | 28 +- crates/tranquil-db/src/postgres/infra.rs | 12 +- crates/tranquil-pds/tests/store_parity.rs | 80 ++++++ .../tranquil-store/src/metastore/blob_ops.rs | 266 +++++++++--------- crates/tranquil-store/src/metastore/blobs.rs | 31 ++ crates/tranquil-store/src/metastore/mod.rs | 45 ++- .../20260906_blob_ownership_per_user.sql | 2 + 15 files changed, 361 insertions(+), 194 deletions(-) rename .sqlx/{query-dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b.json => query-03f129e4984e1bed9e87294adc9caf1730906d889101b9039113ec8aa234618d.json} (84%) rename .sqlx/{query-0890b2c7c921005f58ed0e57b6e062b2085ce804a4cccb27b4ae2ba6711f24c4.json => query-155efbae4cd55f73ec0709dda7b18a76e92065e6ae4a6081bd38a19821fbfcc3.json} (67%) rename .sqlx/{query-62942bd21d545eb15bfea4f46378b6c2ebfe12b8bc9e27c63a6c0f77a9105303.json => query-5996484ff0f8dbc3b278cfd01b8375dbf7bf6da8d903145b12871dda6e1fd5d9.json} (67%) create mode 100644 .sqlx/query-8844d942ef2810afc386e5a9838624ee07a43c380d2df31efdba5cf299aab571.json rename .sqlx/{query-8afea2b745385348f4c78b51f74145d6718bfcf9a3a0c218109ec691aeb930ba.json => query-996e5513fb55670fe3304a6046381e377da6a187dfa3347bd285078a7b4410f2.json} (74%) rename .sqlx/{query-6131bb5b39ca81bdbb193c0a9867bead8d9f3d793ad4eca97a79d166467a5052.json => query-9fb9e128076b20ff067d01955221488ce7e5b886dba0529fb073c3e0461fe030.json} (65%) delete mode 100644 .sqlx/query-f59010ecdd7f782489e0e03288a06dacd72b33d04c1e2b98475018ad25485852.json create mode 100644 migrations/20260906_blob_ownership_per_user.sql diff --git a/.sqlx/query-dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b.json b/.sqlx/query-03f129e4984e1bed9e87294adc9caf1730906d889101b9039113ec8aa234618d.json similarity index 84% rename from .sqlx/query-dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b.json rename to .sqlx/query-03f129e4984e1bed9e87294adc9caf1730906d889101b9039113ec8aa234618d.json index 236230f..910d998 100644 --- a/.sqlx/query-dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b.json +++ b/.sqlx/query-03f129e4984e1bed9e87294adc9caf1730906d889101b9039113ec8aa234618d.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1", + "query": "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1 LIMIT 1", "describe": { "columns": [ { @@ -30,5 +30,5 @@ false ] }, - "hash": "dd1b61d6ec81fd891d4effd3b51e6c22308b878acdc5355dfcb04c5664c9463b" + "hash": "03f129e4984e1bed9e87294adc9caf1730906d889101b9039113ec8aa234618d" } diff --git a/.sqlx/query-0890b2c7c921005f58ed0e57b6e062b2085ce804a4cccb27b4ae2ba6711f24c4.json b/.sqlx/query-155efbae4cd55f73ec0709dda7b18a76e92065e6ae4a6081bd38a19821fbfcc3.json similarity index 67% rename from .sqlx/query-0890b2c7c921005f58ed0e57b6e062b2085ce804a4cccb27b4ae2ba6711f24c4.json rename to .sqlx/query-155efbae4cd55f73ec0709dda7b18a76e92065e6ae4a6081bd38a19821fbfcc3.json index 14db6c2..3f02003 100644 --- a/.sqlx/query-0890b2c7c921005f58ed0e57b6e062b2085ce804a4cccb27b4ae2ba6711f24c4.json +++ b/.sqlx/query-155efbae4cd55f73ec0709dda7b18a76e92065e6ae4a6081bd38a19821fbfcc3.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as \"total!\" FROM blobs", + "query": "SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as \"total!\"\n FROM (SELECT DISTINCT cid, size_bytes FROM blobs) t", "describe": { "columns": [ { @@ -16,5 +16,5 @@ null ] }, - "hash": "0890b2c7c921005f58ed0e57b6e062b2085ce804a4cccb27b4ae2ba6711f24c4" + "hash": "155efbae4cd55f73ec0709dda7b18a76e92065e6ae4a6081bd38a19821fbfcc3" } diff --git a/.sqlx/query-62942bd21d545eb15bfea4f46378b6c2ebfe12b8bc9e27c63a6c0f77a9105303.json b/.sqlx/query-5996484ff0f8dbc3b278cfd01b8375dbf7bf6da8d903145b12871dda6e1fd5d9.json similarity index 67% rename from .sqlx/query-62942bd21d545eb15bfea4f46378b6c2ebfe12b8bc9e27c63a6c0f77a9105303.json rename to .sqlx/query-5996484ff0f8dbc3b278cfd01b8375dbf7bf6da8d903145b12871dda6e1fd5d9.json index f3ae0be..df39367 100644 --- a/.sqlx/query-62942bd21d545eb15bfea4f46378b6c2ebfe12b8bc9e27c63a6c0f77a9105303.json +++ b/.sqlx/query-5996484ff0f8dbc3b278cfd01b8375dbf7bf6da8d903145b12871dda6e1fd5d9.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT cid, takedown_ref FROM blobs WHERE cid = $1", + "query": "SELECT cid, takedown_ref FROM blobs WHERE cid = $1 ORDER BY takedown_ref NULLS LAST LIMIT 1", "describe": { "columns": [ { @@ -24,5 +24,5 @@ true ] }, - "hash": "62942bd21d545eb15bfea4f46378b6c2ebfe12b8bc9e27c63a6c0f77a9105303" + "hash": "5996484ff0f8dbc3b278cfd01b8375dbf7bf6da8d903145b12871dda6e1fd5d9" } diff --git a/.sqlx/query-8844d942ef2810afc386e5a9838624ee07a43c380d2df31efdba5cf299aab571.json b/.sqlx/query-8844d942ef2810afc386e5a9838624ee07a43c380d2df31efdba5cf299aab571.json new file mode 100644 index 0000000..dfc4e45 --- /dev/null +++ b/.sqlx/query-8844d942ef2810afc386e5a9838624ee07a43c380d2df31efdba5cf299aab571.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT storage_key as \"storage_key!\" FROM blobs b\n WHERE created_by_user = $1\n AND NOT EXISTS (\n SELECT 1 FROM blobs o\n WHERE o.cid = b.cid AND o.created_by_user <> $1\n )", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "storage_key!", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false + ] + }, + "hash": "8844d942ef2810afc386e5a9838624ee07a43c380d2df31efdba5cf299aab571" +} diff --git a/.sqlx/query-8afea2b745385348f4c78b51f74145d6718bfcf9a3a0c218109ec691aeb930ba.json b/.sqlx/query-996e5513fb55670fe3304a6046381e377da6a187dfa3347bd285078a7b4410f2.json similarity index 74% rename from .sqlx/query-8afea2b745385348f4c78b51f74145d6718bfcf9a3a0c218109ec691aeb930ba.json rename to .sqlx/query-996e5513fb55670fe3304a6046381e377da6a187dfa3347bd285078a7b4410f2.json index 026d3ff..7857cc0 100644 --- a/.sqlx/query-8afea2b745385348f4c78b51f74145d6718bfcf9a3a0c218109ec691aeb930ba.json +++ b/.sqlx/query-996e5513fb55670fe3304a6046381e377da6a187dfa3347bd285078a7b4410f2.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key)\n VALUES ($1, $2, $3, $4, $5)\n ON CONFLICT (cid) DO NOTHING RETURNING cid", + "query": "INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key)\n VALUES ($1, $2, $3, $4, $5)\n ON CONFLICT (cid, created_by_user) DO NOTHING RETURNING cid", "describe": { "columns": [ { @@ -22,5 +22,5 @@ false ] }, - "hash": "8afea2b745385348f4c78b51f74145d6718bfcf9a3a0c218109ec691aeb930ba" + "hash": "996e5513fb55670fe3304a6046381e377da6a187dfa3347bd285078a7b4410f2" } diff --git a/.sqlx/query-6131bb5b39ca81bdbb193c0a9867bead8d9f3d793ad4eca97a79d166467a5052.json b/.sqlx/query-9fb9e128076b20ff067d01955221488ce7e5b886dba0529fb073c3e0461fe030.json similarity index 65% rename from .sqlx/query-6131bb5b39ca81bdbb193c0a9867bead8d9f3d793ad4eca97a79d166467a5052.json rename to .sqlx/query-9fb9e128076b20ff067d01955221488ce7e5b886dba0529fb073c3e0461fe030.json index 5c07cbf..d05f896 100644 --- a/.sqlx/query-6131bb5b39ca81bdbb193c0a9867bead8d9f3d793ad4eca97a79d166467a5052.json +++ b/.sqlx/query-9fb9e128076b20ff067d01955221488ce7e5b886dba0529fb073c3e0461fe030.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT storage_key FROM blobs WHERE cid = $1", + "query": "SELECT storage_key FROM blobs WHERE cid = $1 LIMIT 1", "describe": { "columns": [ { @@ -18,5 +18,5 @@ false ] }, - "hash": "6131bb5b39ca81bdbb193c0a9867bead8d9f3d793ad4eca97a79d166467a5052" + "hash": "9fb9e128076b20ff067d01955221488ce7e5b886dba0529fb073c3e0461fe030" } diff --git a/.sqlx/query-f59010ecdd7f782489e0e03288a06dacd72b33d04c1e2b98475018ad25485852.json b/.sqlx/query-f59010ecdd7f782489e0e03288a06dacd72b33d04c1e2b98475018ad25485852.json deleted file mode 100644 index 218f7cd..0000000 --- a/.sqlx/query-f59010ecdd7f782489e0e03288a06dacd72b33d04c1e2b98475018ad25485852.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT storage_key as \"storage_key!\" FROM blobs WHERE created_by_user = $1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "storage_key!", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Uuid" - ] - }, - "nullable": [ - false - ] - }, - "hash": "f59010ecdd7f782489e0e03288a06dacd72b33d04c1e2b98475018ad25485852" -} diff --git a/crates/tranquil-api/src/repo/blob.rs b/crates/tranquil-api/src/repo/blob.rs index aa0f7b1..1e2df0f 100644 --- a/crates/tranquil-api/src/repo/blob.rs +++ b/crates/tranquil-api/src/repo/blob.rs @@ -148,7 +148,13 @@ pub async fn upload_blob( size, cid_str ); - match state + if let Err(e) = state.blob_store.copy(&temp_key, &storage_key).await { + let _ = state.blob_store.delete(&temp_key).await; + error!("Failed to copy blob to final location: {:?}", e); + return Err(ApiError::InternalError(Some("Failed to store blob".into()))); + } + + if let Err(e) = state .repos .blob .insert_blob( @@ -160,24 +166,9 @@ pub async fn upload_blob( ) .await { - Ok(_) => {} - Err(e) => { - let _ = state.blob_store.delete(&temp_key).await; - error!("Failed to insert blob record: {:?}", e); - return Err(ApiError::InternalError(None)); - } - }; - - if let Err(e) = state.blob_store.copy(&temp_key, &storage_key).await { let _ = state.blob_store.delete(&temp_key).await; - if let Err(db_err) = state.repos.blob.delete_blob_by_cid(&cid_link).await { - error!( - "Failed to clean up orphaned blob record after copy failure: {:?}", - db_err - ); - } - error!("Failed to copy blob to final location: {:?}", e); - return Err(ApiError::InternalError(Some("Failed to store blob".into()))); + error!("Failed to insert blob record: {:?}", e); + return Err(ApiError::InternalError(None)); } let _ = state.blob_store.delete(&temp_key).await; diff --git a/crates/tranquil-db/src/postgres/blob.rs b/crates/tranquil-db/src/postgres/blob.rs index 6e69a83..f0a1f48 100644 --- a/crates/tranquil-db/src/postgres/blob.rs +++ b/crates/tranquil-db/src/postgres/blob.rs @@ -33,7 +33,7 @@ impl BlobRepository for PostgresBlobRepository { let result = sqlx::query_scalar!( r#"INSERT INTO blobs (cid, mime_type, size_bytes, created_by_user, storage_key) VALUES ($1, $2, $3, $4, $5) - ON CONFLICT (cid) DO NOTHING RETURNING cid"#, + ON CONFLICT (cid, created_by_user) DO NOTHING RETURNING cid"#, cid.as_str(), mime_type, size_bytes, @@ -49,7 +49,7 @@ impl BlobRepository for PostgresBlobRepository { async fn get_blob_metadata(&self, cid: &CidLink) -> Result, DbError> { let result = sqlx::query!( - "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1", + "SELECT storage_key, mime_type, size_bytes FROM blobs WHERE cid = $1 LIMIT 1", cid.as_str() ) .fetch_optional(&self.pool) @@ -68,7 +68,7 @@ impl BlobRepository for PostgresBlobRepository { cid: &CidLink, ) -> Result, DbError> { let result = sqlx::query!( - "SELECT cid, takedown_ref FROM blobs WHERE cid = $1", + "SELECT cid, takedown_ref FROM blobs WHERE cid = $1 ORDER BY takedown_ref NULLS LAST LIMIT 1", cid.as_str() ) .fetch_optional(&self.pool) @@ -86,11 +86,13 @@ impl BlobRepository for PostgresBlobRepository { } async fn get_blob_storage_key(&self, cid: &CidLink) -> Result, DbError> { - let result = - sqlx::query_scalar!("SELECT storage_key FROM blobs WHERE cid = $1", cid.as_str()) - .fetch_optional(&self.pool) - .await - .map_err(map_sqlx_error)?; + let result = sqlx::query_scalar!( + "SELECT storage_key FROM blobs WHERE cid = $1 LIMIT 1", + cid.as_str() + ) + .fetch_optional(&self.pool) + .await + .map_err(map_sqlx_error)?; Ok(result) } @@ -147,7 +149,8 @@ impl BlobRepository for PostgresBlobRepository { async fn sum_blob_storage(&self) -> Result { let result = sqlx::query_scalar!( - r#"SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as "total!" FROM blobs"# + r#"SELECT COALESCE(SUM(size_bytes), 0)::BIGINT as "total!" + FROM (SELECT DISTINCT cid, size_bytes FROM blobs) t"# ) .fetch_one(&self.pool) .await @@ -193,7 +196,12 @@ impl BlobRepository for PostgresBlobRepository { async fn get_blob_storage_keys_by_user(&self, user_id: Uuid) -> Result, DbError> { let results = sqlx::query_scalar!( - r#"SELECT storage_key as "storage_key!" FROM blobs WHERE created_by_user = $1"#, + r#"SELECT storage_key as "storage_key!" FROM blobs b + WHERE created_by_user = $1 + AND NOT EXISTS ( + SELECT 1 FROM blobs o + WHERE o.cid = b.cid AND o.created_by_user <> $1 + )"#, user_id ) .fetch_all(&self.pool) diff --git a/crates/tranquil-db/src/postgres/infra.rs b/crates/tranquil-db/src/postgres/infra.rs index 2ee3ec1..445cbae 100644 --- a/crates/tranquil-db/src/postgres/infra.rs +++ b/crates/tranquil-db/src/postgres/infra.rs @@ -1011,11 +1011,13 @@ impl InfraRepository for PostgresInfraRepository { } async fn get_blob_storage_key_by_cid(&self, cid: &CidLink) -> Result, DbError> { - let result = - sqlx::query_scalar!("SELECT storage_key FROM blobs WHERE cid = $1", cid.as_str()) - .fetch_optional(&self.pool) - .await - .map_err(map_sqlx_error)?; + let result = sqlx::query_scalar!( + "SELECT storage_key FROM blobs WHERE cid = $1 LIMIT 1", + cid.as_str() + ) + .fetch_optional(&self.pool) + .await + .map_err(map_sqlx_error)?; Ok(result) } diff --git a/crates/tranquil-pds/tests/store_parity.rs b/crates/tranquil-pds/tests/store_parity.rs index 42334fc..fc1f656 100644 --- a/crates/tranquil-pds/tests/store_parity.rs +++ b/crates/tranquil-pds/tests/store_parity.rs @@ -986,6 +986,86 @@ async fn parity_blob_duplicate_insert() { assert_eq!(pg_dup, store_dup); } +#[tokio::test(flavor = "multi_thread")] +async fn parity_blob_shared_between_repos() { + let f = ParityFixture::new().await; + + let did_a = test_did("shareda"); + let did_b = test_did("sharedb"); + let (pg_a, store_a) = seed_repos(&f, &did_a, &test_handle("shareda")).await; + let (pg_b, store_b) = seed_repos(&f, &did_b, &test_handle("sharedb")).await; + + let cid = test_cid(210); + + let pg_first = + f.pg.blob + .insert_blob(&cid, "image/png", 100, pg_a, "blobs/shared.png") + .await + .unwrap(); + let store_first = f + .store + .blob + .insert_blob(&cid, "image/png", 100, store_a, "blobs/shared.png") + .await + .unwrap(); + assert_eq!(pg_first, store_first); + + let pg_second = + f.pg.blob + .insert_blob(&cid, "image/png", 100, pg_b, "blobs/shared.png") + .await + .unwrap(); + let store_second = f + .store + .blob + .insert_blob(&cid, "image/png", 100, store_b, "blobs/shared.png") + .await + .unwrap(); + assert_eq!(pg_second, store_second); + assert!(pg_second.is_some()); + + for (pg_uid, store_uid) in [(pg_a, store_a), (pg_b, store_b)] { + assert_eq!(f.pg.blob.count_blobs_by_user(pg_uid).await.unwrap(), 1); + assert_eq!( + f.store.blob.count_blobs_by_user(store_uid).await.unwrap(), + 1 + ); + assert_eq!( + f.pg.blob + .list_blobs_by_user(pg_uid, None, 100) + .await + .unwrap(), + vec![cid.clone()] + ); + assert_eq!( + f.store + .blob + .list_blobs_by_user(store_uid, None, 100) + .await + .unwrap(), + vec![cid.clone()] + ); + assert!( + f.pg.blob + .get_blob_storage_keys_by_user(pg_uid) + .await + .unwrap() + .is_empty() + ); + assert!( + f.store + .blob + .get_blob_storage_keys_by_user(store_uid) + .await + .unwrap() + .is_empty() + ); + } + + assert_eq!(f.pg.blob.sum_blob_storage().await.unwrap(), 100); + assert_eq!(f.store.blob.sum_blob_storage().await.unwrap(), 100); +} + #[tokio::test] async fn parity_get_all_records() { let f = ParityFixture::new().await; diff --git a/crates/tranquil-store/src/metastore/blob_ops.rs b/crates/tranquil-store/src/metastore/blob_ops.rs index 50c496f..69f9165 100644 --- a/crates/tranquil-store/src/metastore/blob_ops.rs +++ b/crates/tranquil-store/src/metastore/blob_ops.rs @@ -7,7 +7,10 @@ use smallvec::SmallVec; use uuid::Uuid; use super::MetastoreError; -use super::blobs::{BlobMetaValue, blob_by_cid_key, blob_meta_key, blob_user_prefix, blobs_prefix}; +use super::blobs::{ + BlobContentValue, BlobMetaValue, blob_by_cid_key, blob_by_cid_prefix, blob_meta_key, + blob_user_prefix, +}; use super::commit_ops::{RecordBlobsValue, record_blobs_user_prefix}; use super::encoding::{KeyReader, exclusive_upper_bound}; use super::keys::{KeyTag, UserHash}; @@ -55,69 +58,62 @@ impl BlobOps { let user_hash = self.resolve_user_hash(created_by_user)?; let cid_str = cid.as_str(); - - let cid_index_key = blob_by_cid_key(cid_str); - let existing = self + let marker_key = blob_meta_key(user_hash, cid_str); + if self .repo_data - .get(cid_index_key.as_slice()) - .map_err(MetastoreError::Fjall)?; - if existing.is_some() { + .get(marker_key.as_slice()) + .map_err(MetastoreError::Fjall)? + .is_some() + { return Ok(None); } - let value = BlobMetaValue { - size_bytes, - mime_type: mime_type.to_owned(), - storage_key: storage_key.to_owned(), - takedown_ref: None, - created_at_ms: chrono::Utc::now().timestamp_millis(), + let cid_index_key = blob_by_cid_key(cid_str); + let content = match point_lookup( + &self.repo_data, + cid_index_key.as_slice(), + BlobContentValue::deserialize, + "corrupt blob_content value", + )? { + Some(mut existing) => { + existing.ref_count = existing.ref_count.saturating_add(1); + existing + } + None => BlobContentValue { + meta: BlobMetaValue { + size_bytes, + mime_type: mime_type.to_owned(), + storage_key: storage_key.to_owned(), + takedown_ref: None, + created_at_ms: chrono::Utc::now().timestamp_millis(), + }, + ref_count: 1, + }, }; - let primary_key = blob_meta_key(user_hash, cid_str); - let mut batch = self.db.batch(); - batch.insert(&self.repo_data, primary_key.as_slice(), value.serialize()); + batch.insert(&self.repo_data, marker_key.as_slice(), &[] as &[u8]); batch.insert( &self.repo_data, cid_index_key.as_slice(), - user_hash.raw().to_be_bytes(), + content.serialize(), ); batch.commit().map_err(MetastoreError::Fjall)?; Ok(Some(cid.clone())) } - fn lookup_user_hash_by_cid(&self, cid_str: &str) -> Result, MetastoreError> { - let key = blob_by_cid_key(cid_str); - match self - .repo_data - .get(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"))?; - Ok(Some(UserHash::from_raw(u64::from_be_bytes(arr)))) - } - None => Ok(None), - } + fn get_blob_content(&self, cid: &CidLink) -> Result, MetastoreError> { + point_lookup( + &self.repo_data, + blob_by_cid_key(cid.as_str()).as_slice(), + BlobContentValue::deserialize, + "corrupt blob_content value", + ) } fn get_blob_value(&self, cid: &CidLink) -> Result, MetastoreError> { - let cid_str = cid.as_str(); - let user_hash = match self.lookup_user_hash_by_cid(cid_str)? { - Some(h) => h, - None => return Ok(None), - }; - let key = blob_meta_key(user_hash, cid_str); - point_lookup( - &self.repo_data, - key.as_slice(), - BlobMetaValue::deserialize, - "corrupt blob_meta value", - ) + Ok(self.get_blob_content(cid)?.map(|c| c.meta)) } pub fn get_blob_metadata( @@ -186,14 +182,14 @@ impl BlobOps { } pub fn sum_blob_storage(&self) -> Result { - let prefix = blobs_prefix(); + let prefix = blob_by_cid_prefix(); self.repo_data .prefix(prefix.as_slice()) .try_fold(0i64, |acc, guard| { let (_, val_bytes) = guard.into_inner().map_err(MetastoreError::Fjall)?; - let value = BlobMetaValue::deserialize(&val_bytes) - .ok_or(MetastoreError::CorruptData("corrupt blob_meta in sum"))?; - Ok::<_, MetastoreError>(acc.saturating_add(value.size_bytes)) + let content = BlobContentValue::deserialize(&val_bytes) + .ok_or(MetastoreError::CorruptData("corrupt blob_content in sum"))?; + Ok::<_, MetastoreError>(acc.saturating_add(content.meta.size_bytes)) }) } @@ -202,50 +198,34 @@ impl BlobOps { cid: &CidLink, takedown_ref: Option<&str>, ) -> Result { - let cid_str = cid.as_str(); - let user_hash = match self.lookup_user_hash_by_cid(cid_str)? { - Some(h) => h, - None => return Ok(false), - }; - let key = blob_meta_key(user_hash, cid_str); - let mut value = match point_lookup( - &self.repo_data, - key.as_slice(), - BlobMetaValue::deserialize, - "corrupt blob_meta value", - )? { - Some(v) => v, + let mut content = match self.get_blob_content(cid)? { + Some(c) => c, None => return Ok(false), }; - value.takedown_ref = takedown_ref.map(str::to_owned); + content.meta.takedown_ref = takedown_ref.map(str::to_owned); let mut batch = self.db.batch(); - batch.insert(&self.repo_data, key.as_slice(), value.serialize()); + batch.insert( + &self.repo_data, + blob_by_cid_key(cid.as_str()).as_slice(), + content.serialize(), + ); batch.commit().map_err(MetastoreError::Fjall)?; Ok(true) } pub fn delete_blob_by_cid(&self, cid: &CidLink) -> Result { - let cid_str = cid.as_str(); - let user_hash = match self.lookup_user_hash_by_cid(cid_str)? { - Some(h) => h, - None => return Ok(false), - }; - - let primary_key = blob_meta_key(user_hash, cid_str); - let exists = self + let cid_index_key = blob_by_cid_key(cid.as_str()); + if self .repo_data - .get(primary_key.as_slice()) + .get(cid_index_key.as_slice()) .map_err(MetastoreError::Fjall)? - .is_some(); - if !exists { + .is_none() + { return Ok(false); } - let cid_index_key = blob_by_cid_key(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)?; @@ -255,7 +235,6 @@ impl BlobOps { pub fn delete_blobs_by_user(&self, user_id: Uuid) -> Result { let user_hash = self.resolve_user_hash(user_id)?; let prefix = blob_user_prefix(user_hash); - let user_hash_bytes = user_hash.raw().to_be_bytes(); let (final_batch, remaining, total) = self .repo_data @@ -273,14 +252,25 @@ impl BlobOps { blob_meta_key(user_hash, &cid_str).as_slice(), ); let cid_index_key = blob_by_cid_key(&cid_str); - let owns_cid = self - .repo_data - .get(cid_index_key.as_slice()) - .map_err(MetastoreError::Fjall)? - .is_some_and(|raw| raw.as_ref() == user_hash_bytes); - if owns_cid { - batch.remove(&self.repo_data, cid_index_key.as_slice()); + + if let Some(mut content) = point_lookup( + &self.repo_data, + cid_index_key.as_slice(), + BlobContentValue::deserialize, + "corrupt blob_content value", + )? { + content.ref_count = content.ref_count.saturating_sub(1); + if content.ref_count == 0 { + batch.remove(&self.repo_data, cid_index_key.as_slice()); + } else { + batch.insert( + &self.repo_data, + cid_index_key.as_slice(), + content.serialize(), + ); + } } + let new_count = count + 1; if new_count >= DELETE_BATCH_SIZE { batch.commit().map_err(MetastoreError::Fjall)?; @@ -311,11 +301,14 @@ impl BlobOps { self.repo_data .prefix(prefix.as_slice()) .map(|guard| { - let (_, val_bytes) = guard.into_inner().map_err(MetastoreError::Fjall)?; - let value = BlobMetaValue::deserialize(&val_bytes) - .ok_or(MetastoreError::CorruptData("corrupt blob_meta value"))?; - Ok(value.storage_key) + let (key_bytes, _) = guard.into_inner().map_err(MetastoreError::Fjall)?; + let cid = parse_blob_cid_from_key(key_bytes.as_ref())?; + Ok(self + .get_blob_content(&cid)? + .filter(|c| c.ref_count == 1) + .map(|c| c.meta.storage_key)) }) + .filter_map(Result::transpose) .collect() } @@ -346,12 +339,7 @@ impl BlobOps { if acc.contains_key(&cid_str) { return Ok(()); } - let key = blob_meta_key(user_hash, &cid_str); - let exists = self - .repo_data - .get(key.as_slice()) - .map_err(MetastoreError::Fjall)? - .is_some(); + let exists = self.get_blob_content(&cid_link)?.is_some(); if !exists { acc.insert(cid_str, record_uri.clone()); } @@ -415,17 +403,11 @@ impl BlobOps { Ok(c) => c, Err(e) => return Some(Err(e)), }; - let key = blob_meta_key(user_hash, cid_link.as_str()); - match point_lookup( - &self.repo_data, - key.as_slice(), - BlobMetaValue::deserialize, - "corrupt blob_meta value", - ) { - Ok(Some(v)) => Some(Ok(tranquil_db_traits::BlobForExport { + match self.get_blob_content(&cid_link) { + Ok(Some(c)) => Some(Ok(tranquil_db_traits::BlobForExport { cid: cid_link, - storage_key: v.storage_key, - mime_type: v.mime_type, + storage_key: c.meta.storage_key, + mime_type: c.meta.mime_type, })), Ok(None) => None, Err(e) => Some(Err(e)), @@ -537,26 +519,6 @@ mod tests { ); } - #[test] - fn insert_same_cid_different_user_returns_none() { - let (_dir, ms) = open_fresh(); - let (user_a, _) = setup_user(&ms); - let (user_b, _) = setup_user(&ms); - let ops = ms.blob_ops(); - - let cid = test_cid_link(80); - assert!( - ops.insert_blob(&cid, "image/png", 100, user_a, "ka") - .unwrap() - .is_some() - ); - assert!( - ops.insert_blob(&cid, "image/png", 100, user_b, "kb") - .unwrap() - .is_none() - ); - } - #[test] fn get_blob_with_takedown_no_takedown() { let (_dir, ms) = open_fresh(); @@ -704,7 +666,7 @@ mod tests { ops.delete_blob_by_cid(&cid).unwrap(); - assert!(ops.lookup_user_hash_by_cid(cid.as_str()).unwrap().is_none()); + assert!(ops.get_blob_metadata(&cid).unwrap().is_none()); } #[test] @@ -735,7 +697,7 @@ mod tests { ops.delete_blobs_by_user(user_id).unwrap(); - assert!(ops.lookup_user_hash_by_cid(cid.as_str()).unwrap().is_none()); + assert!(ops.get_blob_metadata(&cid).unwrap().is_none()); } #[test] @@ -787,4 +749,52 @@ mod tests { assert_eq!(ops.count_blobs_by_user(user_b).unwrap(), 1); assert_eq!(ops.sum_blob_storage().unwrap(), 30); } + + #[test] + fn blob_shared_between_users() { + let (_dir, ms) = open_fresh(); + let (user_a, _) = setup_user(&ms); + let (user_b, _) = setup_user(&ms); + let ops = ms.blob_ops(); + + let cid = test_cid_link(80); + + assert!( + ops.insert_blob(&cid, "a/b", 10, user_a, "k") + .unwrap() + .is_some() + ); + assert!( + ops.insert_blob(&cid, "a/b", 10, user_b, "k") + .unwrap() + .is_some() + ); + assert!( + ops.insert_blob(&cid, "a/b", 10, user_b, "k") + .unwrap() + .is_none() + ); + + assert_eq!(ops.count_blobs_by_user(user_a).unwrap(), 1); + assert_eq!(ops.count_blobs_by_user(user_b).unwrap(), 1); + assert_eq!( + ops.list_blobs_by_user(user_b, None, 100).unwrap(), + vec![cid.clone()] + ); + + assert_eq!(ops.sum_blob_storage().unwrap(), 10); + assert!( + ops.get_blob_storage_keys_by_user(user_a) + .unwrap() + .is_empty() + ); + + ops.delete_blobs_by_user(user_a).unwrap(); + assert_eq!(ops.count_blobs_by_user(user_b).unwrap(), 1); + assert!(ops.get_blob_metadata(&cid).unwrap().is_some()); + assert_eq!( + ops.get_blob_storage_keys_by_user(user_b).unwrap(), + vec!["k".to_string()] + ); + } } diff --git a/crates/tranquil-store/src/metastore/blobs.rs b/crates/tranquil-store/src/metastore/blobs.rs index 94733ee..798a62c 100644 --- a/crates/tranquil-store/src/metastore/blobs.rs +++ b/crates/tranquil-store/src/metastore/blobs.rs @@ -33,6 +33,33 @@ impl BlobMetaValue { } } +const BLOB_CONTENT_SCHEMA_VERSION: u8 = 1; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct BlobContentValue { + pub meta: BlobMetaValue, + pub ref_count: u32, +} + +impl BlobContentValue { + pub fn serialize(&self) -> Vec { + let payload = + postcard::to_allocvec(self).expect("BlobContentValue serialization cannot fail"); + let mut buf = Vec::with_capacity(1 + payload.len()); + buf.push(BLOB_CONTENT_SCHEMA_VERSION); + buf.extend_from_slice(&payload); + buf + } + + pub fn deserialize(bytes: &[u8]) -> Option { + let (&version, payload) = bytes.split_first()?; + match version { + BLOB_CONTENT_SCHEMA_VERSION => postcard::from_bytes(payload).ok(), + _ => None, + } + } +} + pub fn blob_meta_key(user_hash: UserHash, cid_str: &str) -> SmallVec<[u8; 128]> { KeyBuilder::new() .tag(KeyTag::BLOBS) @@ -59,6 +86,10 @@ pub fn blob_by_cid_key(cid_str: &str) -> SmallVec<[u8; 128]> { .build() } +pub fn blob_by_cid_prefix() -> SmallVec<[u8; 128]> { + KeyBuilder::new().tag(KeyTag::BLOB_BY_CID).build() +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/tranquil-store/src/metastore/mod.rs b/crates/tranquil-store/src/metastore/mod.rs index 50af461..15f382e 100644 --- a/crates/tranquil-store/src/metastore/mod.rs +++ b/crates/tranquil-store/src/metastore/mod.rs @@ -35,11 +35,12 @@ use std::sync::Arc; use fjall::{Database, Keyspace}; +use self::encoding::KeyReader; use self::keys::KeyTag; use self::partitions::Partition; use self::user_hash::UserHashMap; -const CURRENT_FORMAT_VERSION: u64 = 2; +const CURRENT_FORMAT_VERSION: u64 = 3; #[derive(Debug, Clone)] pub struct MetastoreConfig { @@ -240,6 +241,7 @@ impl Metastore { "upgrading metastore format and rebuilding derived indexes" ); repo_data.remove(records::record_by_cid_built_key().as_slice())?; + Self::migrate_blob_ownership(db, repo_data)?; repo_data.insert(version_key, version_bytes)?; db.persist(fjall::PersistMode::SyncData)?; Ok(()) @@ -254,6 +256,47 @@ impl Metastore { } } + fn migrate_blob_ownership(db: &Database, repo_data: &Keyspace) -> Result<(), MetastoreError> { + let entries: Vec<(Vec, Vec)> = repo_data + .prefix(blobs::blobs_prefix().as_slice()) + .map(|guard| { + let (k, v) = guard.into_inner()?; + Ok((k.as_ref().to_vec(), v.as_ref().to_vec())) + }) + .collect::>()?; + + for (key_bytes, val_bytes) in entries { + let Some(meta) = blobs::BlobMetaValue::deserialize(&val_bytes) else { + continue; + }; + let mut reader = KeyReader::new(&key_bytes); + reader.tag(); + reader.u64(); + let Some(cid_str) = reader.string() else { + continue; + }; + + let cid_index_key = blobs::blob_by_cid_key(cid_str.as_str()); + let content = match repo_data + .get(cid_index_key.as_slice())? + .and_then(|raw| blobs::BlobContentValue::deserialize(raw.as_ref())) + { + Some(mut existing) => { + existing.ref_count = existing.ref_count.saturating_add(1); + existing + } + None => blobs::BlobContentValue { meta, ref_count: 1 }, + }; + + let mut batch = db.batch(); + batch.insert(repo_data, cid_index_key.as_slice(), content.serialize()); + batch.insert(repo_data, key_bytes.as_slice(), &[] as &[u8]); + batch.commit()?; + } + + Ok(()) + } + pub fn path(&self) -> &Path { &self.path } diff --git a/migrations/20260906_blob_ownership_per_user.sql b/migrations/20260906_blob_ownership_per_user.sql new file mode 100644 index 0000000..46bcee7 --- /dev/null +++ b/migrations/20260906_blob_ownership_per_user.sql @@ -0,0 +1,2 @@ +ALTER TABLE blobs DROP CONSTRAINT IF EXISTS blobs_pkey; +ALTER TABLE blobs ADD PRIMARY KEY (cid, created_by_user);