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.

.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"
 }

.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"
 }

.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"
 }

.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"
+}

.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"
 }

.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"
 }

.sqlx/query-f59010ecdd7f782489e0e03288a06dacd72b33d04c1e2b98475018ad25485852.json

@@ -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"
-}

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;

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<Option<BlobMetadata>, 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<Option<BlobWithTakedown>, 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<Option<String>, 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<i64, DbError> {
         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<Vec<String>, 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)

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<Option<String>, 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)
     }

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;

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<Option<UserHash>, 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<Option<BlobContentValue>, 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<Option<BlobMetaValue>, 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<i64, MetastoreError> {
-        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<bool, 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(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<bool, 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(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<u64, MetastoreError> {
         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()]
+        );
+    }
 }

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<u8> {
+        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<Self> {
+        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::*;

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<u8>, Vec<u8>)> = 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::<Result<_, fjall::Error>>()?;
+
+        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
     }

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);