fix(emule): allow identical content at multiple paths in the seed index

Twin of the shared_files fix, for emule_shared_files.ed2k_hash. It was
UNIQUE with `upsert ... ON CONFLICT(ed2k_hash) DO UPDATE SET path=...`,
so duplicate-content files didn't error but ping-ponged: the single row
kept getting repointed between the copies, and since the backfill picks
candidates by path, the two paths re-hashed (MD4) each other's slot every
sweep.

Key the seed index by path too:
- Schema: drop UNIQUE on ed2k_hash, add UNIQUE on path, index ed2k_hash.
- upsert conflicts on path (one row per file; re-seeding a path refreshes
  its hash).
- get_by_hash returns the oldest matching row deterministically.
- Replace delete_by_hash (removed all copies) with delete_by_path, which
  reports whether it was the last copy so callers drop the content from
  the upload whitelist only when nothing seeds it anymore.
- The source republisher publishes each hash once per round.

Author: ogarcia

rucio-daemon/src/db/emule_shared_files.rs

@@ -42,11 +42,14 @@ pub async fn upsert(
     hashset: &[u8],
     now: u64,
 ) -> Result<()> {
+    // Keyed by path (one row per file): identical content at two paths gets two
+    // rows, both seeded. Re-seeding the same path refreshes it (the content there
+    // may have changed, so ed2k_hash is updated too).
     sqlx::query(
         "INSERT INTO emule_shared_files (ed2k_hash, name, size, path, mtime, hashset, added_at) \
          VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) \
-         ON CONFLICT(ed2k_hash) DO UPDATE SET \
-             name = excluded.name, size = excluded.size, path = excluded.path, \
+         ON CONFLICT(path) DO UPDATE SET \
+             ed2k_hash = excluded.ed2k_hash, name = excluded.name, size = excluded.size, \
              mtime = excluded.mtime, hashset = excluded.hashset",
     )
     .bind(ed2k_hash.as_slice())
@@ -132,20 +135,40 @@ pub async fn rename_path(db: &Db, old_path: &str, new_path: &str, new_name: &str
     Ok(affected > 0)
 }
 
-/// Remove a shared file by its ed2k hash. Returns `true` if a row was deleted.
-pub async fn delete_by_hash(db: &Db, ed2k_hash: &[u8]) -> Result<bool> {
-    let res = sqlx::query("DELETE FROM emule_shared_files WHERE ed2k_hash = ?1")
-        .bind(ed2k_hash)
+/// Stop seeding the file at exactly `path`. Returns `true` when that was the
+/// **last** path holding the content (no other row shares its ed2k hash), so the
+/// caller also drops it from the upload whitelist / lets its Kad source expire —
+/// a duplicate copy elsewhere keeps it seeded.
+pub async fn delete_by_path(db: &Db, path: &str) -> Result<bool> {
+    let row = sqlx::query("SELECT ed2k_hash FROM emule_shared_files WHERE path = ?1")
+        .bind(path)
+        .fetch_optional(db)
+        .await?;
+
+    let Some(hash) = row.map(|r| r.get::<Vec<u8>, _>("ed2k_hash")) else {
+        return Ok(false);
+    };
+
+    sqlx::query("DELETE FROM emule_shared_files WHERE path = ?1")
+        .bind(path)
         .execute(db)
         .await?;
-    Ok(res.rows_affected() > 0)
+
+    let remaining: i64 =
+        sqlx::query_scalar("SELECT COUNT(*) FROM emule_shared_files WHERE ed2k_hash = ?1")
+            .bind(hash.as_slice())
+            .fetch_one(db)
+            .await?;
+    Ok(remaining == 0)
 }
 
 /// Look up a shared file by its ed2k hash — used to warn the user that content
-/// they're about to download is already present (and where).
+/// they're about to download is already present (and where). Content can be
+/// seeded from several paths; returns the oldest row deterministically.
 pub async fn get_by_hash(db: &Db, ed2k_hash: &[u8]) -> Result<Option<EmuleSharedFile>> {
     let row = sqlx::query(
-        "SELECT ed2k_hash, name, size, path, mtime, hashset FROM emule_shared_files WHERE ed2k_hash = ?1",
+        "SELECT ed2k_hash, name, size, path, mtime, hashset FROM emule_shared_files \
+         WHERE ed2k_hash = ?1 ORDER BY id ASC LIMIT 1",
     )
     .bind(ed2k_hash)
     .fetch_optional(db)
@@ -320,4 +343,31 @@ mod tests {
         assert_eq!(cands.len(), 1);
         assert_eq!(cands[0].path, "/tmp/b.bin");
     }
+
+    #[tokio::test]
+    async fn duplicate_content_at_two_paths_seeds_once() {
+        let (db, _dir) = test_db().await;
+        let hash = [5u8; 16];
+        // Same content, two paths: both seed (no UNIQUE(ed2k_hash) collision, no
+        // path ping-pong).
+        upsert(&db, &hash, "dup", 10, "/dl/a", 1, b"", 1)
+            .await
+            .unwrap();
+        upsert(&db, &hash, "dup", 10, "/dl/b", 1, b"", 2)
+            .await
+            .unwrap();
+        assert_eq!(list(&db).await.unwrap().len(), 2);
+        // get_by_hash resolves deterministically to the oldest row.
+        assert_eq!(
+            get_by_hash(&db, &hash).await.unwrap().unwrap().path,
+            "/dl/a"
+        );
+
+        // Removing one copy is not the last → keep seeding the content.
+        assert!(!delete_by_path(&db, "/dl/a").await.unwrap());
+        assert!(get_by_hash(&db, &hash).await.unwrap().is_some());
+        // Removing the last copy → caller should stop serving it.
+        assert!(delete_by_path(&db, "/dl/b").await.unwrap());
+        assert!(get_by_hash(&db, &hash).await.unwrap().is_none());
+    }
 }

rucio-daemon/src/db/schema.sql

@@ -237,15 +237,20 @@ CREATE INDEX IF NOT EXISTS idx_dl_chunks_status ON download_chunks(download_id,
 -- ---------------------------------------------------------------------------
 CREATE TABLE IF NOT EXISTS emule_shared_files (
     id          INTEGER PRIMARY KEY,
-    ed2k_hash   BLOB    NOT NULL UNIQUE,  -- 16 bytes MD4, canonical identifier
+    ed2k_hash   BLOB    NOT NULL,         -- 16 bytes MD4, content id (NOT unique: identical content may live at several paths)
     name        TEXT    NOT NULL,
     size        INTEGER NOT NULL,
-    path        TEXT    NOT NULL,         -- absolute path of the final file on disk
+    path        TEXT    NOT NULL UNIQUE,  -- absolute path of the final file on disk, the per-row key
     mtime       INTEGER NOT NULL,         -- file mtime in Unix seconds (change signal)
     hashset     BLOB    NOT NULL DEFAULT X'',  -- ed2k part-hash set, 16 bytes per part (empty for single-part files)
     added_at    INTEGER NOT NULL
 );
 
+-- Look up seeded files by content hash (get_by_hash, the republish sweep).
+-- ed2k_hash is no longer UNIQUE, so it needs its own index; the per-path lookups
+-- (backfill join, watcher) use the implicit unique index on path.
+CREATE INDEX IF NOT EXISTS idx_emule_shared_files_ed2k_hash ON emule_shared_files(ed2k_hash);
+
 -- ---------------------------------------------------------------------------
 -- metrics
 -- Single-row table holding cumulative lifetime counters.

rucio-daemon/src/emule.rs

@@ -195,7 +195,7 @@ pub async fn load_shared_files(db: &Db, active_downloads: &ActiveDownloads) {
         let unchanged =
             disk_size == row.size && crate::api::shares::file_mtime_secs(&path) == row.mtime;
         if !unchanged {
-            let _ = crate::db::emule_shared_files::delete_by_hash(db, &row.ed2k_hash).await;
+            let _ = crate::db::emule_shared_files::delete_by_path(db, &row.path).await;
             dropped += 1;
             continue;
         }
@@ -291,8 +291,12 @@ pub fn spawn_shared_files_watcher(
                 if unchanged {
                     continue; // genuine no-op (or our own completion) — keep sharing
                 }
-                let _ = crate::db::emule_shared_files::delete_by_hash(&db, &row.ed2k_hash).await;
-                if let Ok(hash) = <[u8; 16]>::try_from(row.ed2k_hash.as_slice()) {
+                // Drop just this path; only stop serving the content if no other
+                // copy still seeds it.
+                let was_last = crate::db::emule_shared_files::delete_by_path(&db, &path_str)
+                    .await
+                    .unwrap_or(false);
+                if was_last && let Ok(hash) = <[u8; 16]>::try_from(row.ed2k_hash.as_slice()) {
                     active_downloads.write().await.remove(&hash);
                 }
                 info!(path = %path.display(), "eMule shared file changed/removed — stopped sharing");
@@ -326,8 +330,13 @@ async fn handle_emule_rename(
 
     if !unchanged {
         // Content changed alongside the rename → stop seeding the stale file.
-        let _ = crate::db::emule_shared_files::delete_by_hash(db, &row.ed2k_hash).await;
-        active_downloads.write().await.remove(&hash);
+        // Only forget the content if this was its last seeded copy.
+        let was_last = crate::db::emule_shared_files::delete_by_path(db, &old_str)
+            .await
+            .unwrap_or(false);
+        if was_last {
+            active_downloads.write().await.remove(&hash);
+        }
         info!(path = %new.display(), "eMule shared file changed on rename — stopped sharing");
         return;
     }
@@ -424,10 +433,16 @@ pub fn spawn_source_republisher(
                     count = files.len(),
                     "Republishing eMule shared files as Kad sources"
                 );
+                // Identical content seeded from several paths shares one hash;
+                // publish each hash once per round.
+                let mut published = std::collections::HashSet::new();
                 for row in files {
                     let Ok(bytes) = <[u8; 16]>::try_from(row.ed2k_hash.as_slice()) else {
                         continue;
                     };
+                    if !published.insert(bytes) {
+                        continue;
+                    }
                     let hash = rucio_emule::ed2k::Ed2kHash::from_bytes(bytes);
                     let stored = kad.publish_source(hash, row.size.max(0) as u64).await;
                     debug!(