feat(fuse): implement link() as a server-side copy (#206)

## Problem

`link()` returns ENOTSUP, so hardlink-first import pipelines fall back
to a full copy through the mount. Copying a file whose bytes already
live in CAS re-chunks and re-uploads everything, and dedup against the
existing xorbs runs into xet-core's defrag prevention, which re-stores a
large share of the bytes in short fragments.

Measured on production buckets (multi-GB media files imported this way):
~2x stored bytes, +90-160% xorb overhead, reconstructions of 15-17k
segments alternating between two xorb sets, CPR pinned at the
defrag-prevention hysteresis equilibrium (~8 chunks/segment). Related:
the metrics side of that investigation is huggingface/xet-core#886.

## Change

Implement `link()` as a **server-side copy**: one batch `AddFile`
pointing at the source's committed xet hash. No bytes move through CAS,
the import is instant, and the original clean layout is preserved.

The alias gets its own inode, so this intentionally diverges from POSIX
same-inode semantics: writes to one path never affect the other, and
`st_ino` differs between the two paths. Callers that hardlink for
instant-copy semantics (the *arr import case) get exactly what they
need. The previous ENOTSUP rationale (in-memory links never persisted to
the Hub) no longer applies since the alias is a real Hub entry.

Guards, mirroring the rename/create paths:
- dirty source or no committed hash: ENOTSUP (callers keep their copy
fallback, and we never alias a stale hash)
- overlay mode: ENOTSUP (must not mutate the remote)
- directory source: EPERM, missing parent: ENOENT, non-directory parent:
ENOTDIR, existing target: EEXIST (checked against remote children too,
and re-checked under the write lock)
- read-only mount: EROFS, OS junk names: EACCES

Structured like `rename()`: validate under read lock, commit the batch
op, insert the alias under write lock (with negative-cache removal and
queued-delete cancellation for the destination path).

NFS backend unchanged: nfsserve 0.11 does not dispatch `NFSPROC3_LINK`
(proc_unavail), so hardlinks over NFS fail at the protocol level before
reaching us. Supporting it would need an upstream nfsserve change.

## Tests

Four new tests in `virtual_fs/tests.rs`: happy path (exactly one AddFile
with the source hash, no delete, alias resolvable with its own inode,
source untouched), dirty source (ENOTSUP, no remote op), existing target
(EEXIST, no remote op), directory source (EPERM). Full lib suite: 340
passed.

Author: XciD

src/fuse.rs

@@ -417,8 +417,16 @@ impl Filesystem for FuseAdapter {
         }
     }
 
-    fn link(&self, _req: &Request, _ino: INodeNo, _newparent: INodeNo, _newname: &OsStr, reply: ReplyEntry) {
-        reply.error(Errno::from_i32(libc::ENOTSUP));
+    /// Hard link, implemented as a server-side copy: a new Hub entry pointing at
+    /// the source's xet hash (no data re-uploaded). The alias gets its own inode,
+    /// so st_ino differs from the source; callers like the *arr import pipelines
+    /// only need the instant copy semantics.
+    fn link(&self, _req: &Request, ino: INodeNo, newparent: INodeNo, newname: &OsStr, reply: ReplyEntry) {
+        let newname = os_to_str!(newname, reply);
+        match self.runtime.block_on(self.virtual_fs.link(ino.0, newparent.0, newname)) {
+            Ok(attr) => self.reply_entry_tracked(reply, &attr),
+            Err(e) => reply.error(Errno::from_i32(e)),
+        }
     }
 
     /// Remove an empty directory.

src/virtual_fs/mod.rs

@@ -2928,6 +2928,130 @@ impl VirtualFs {
         Ok(self.make_vfs_attr(inodes.get(ino).ok_or(libc::ENOENT)?))
     }
 
+    /// Hard-link `ino` at `newparent/newname` as a server-side copy: a new Hub
+    /// file entry referencing the source's xet hash, so no bytes move through
+    /// CAS. The alias gets its own inode (writes to one path never affect the
+    /// other), which diverges from POSIX same-inode semantics but matches what
+    /// link() callers like the *arr import pipelines need: an instant,
+    /// storage-free copy. Re-uploading the content instead would shred its
+    /// layout (dedup against the existing xorbs runs into xet-core's defrag
+    /// prevention, re-storing a large share of the bytes in fragments).
+    ///
+    /// Sources without a committed hash (dirty, or created and never flushed)
+    /// return ENOTSUP so callers fall back to a regular copy.
+    pub async fn link(&self, ino: u64, newparent: u64, newname: &str) -> VirtualFsResult<VirtualFsAttr> {
+        if self.read_only {
+            return Err(libc::EROFS);
+        }
+        // Overlay keeps writes local; committing an alias to the Hub would
+        // mutate the remote behind the overlay's back.
+        if self.overlay() {
+            return Err(libc::ENOTSUP);
+        }
+        if self.filter_os_files && is_os_junk(newname) {
+            debug!("link: rejecting OS junk name: {}", newname);
+            return Err(libc::EACCES);
+        }
+
+        debug!("link: ino={}, newparent={}, newname={}", ino, newparent, newname);
+
+        // Phase 1a: validate the source under a read lock, before loading the
+        // destination's remote children. A rejected source (dirty or hashless,
+        // the routine *arr link-then-copy fallback) must not pay for a
+        // list_tree it would throw away.
+        let (source_path, xet_hash, size, mode, uid, gid) = {
+            let inodes = self.inode_table.read().expect("inodes poisoned");
+            let source = inodes.get(ino).ok_or(libc::ENOENT)?;
+            if source.kind != InodeKind::File {
+                return Err(libc::EPERM);
+            }
+            if source.is_dirty() {
+                return Err(libc::ENOTSUP);
+            }
+            let Some(hash) = source.xet_hash.as_ref().filter(|hash| !hash.is_empty()).cloned() else {
+                return Err(libc::ENOTSUP);
+            };
+            (
+                source.full_path.clone(),
+                hash,
+                source.size,
+                source.mode,
+                source.uid,
+                source.gid,
+            )
+        };
+
+        // Load remote children so the EEXIST check below also sees files that
+        // exist remotely but were never looked up (also validates newparent).
+        self.ensure_children_loaded(newparent).await?;
+
+        // Phase 1b: validate the destination under a read lock.
+        let new_full_path = {
+            let inodes = self.inode_table.read().expect("inodes poisoned");
+            let parent_entry = match inodes.get(newparent) {
+                Some(e) if e.kind == InodeKind::Directory => e,
+                Some(_) => return Err(libc::ENOTDIR),
+                None => return Err(libc::ENOENT),
+            };
+            let new_full_path = inode::child_path(&parent_entry.full_path, newname);
+            if inodes.lookup_child(newparent, newname).is_some() {
+                return Err(libc::EEXIST);
+            }
+            new_full_path
+        };
+
+        // Phase 2: commit the alias to the Hub.
+        let now = SystemTime::now();
+        let mtime_ms = now.duration_since(UNIX_EPOCH).unwrap_or_default().as_millis() as u64;
+        let ops = [BatchOp::AddFile {
+            path: new_full_path.clone(),
+            xet_hash: xet_hash.clone(),
+            mtime: mtime_ms,
+            content_type: None,
+        }];
+        if let Err(e) = self.hub_client.batch_operations(&ops).await {
+            error!(
+                "link: failed to commit {} as alias of {}: {}",
+                new_full_path, source_path, e
+            );
+            return Err(libc::EIO);
+        }
+
+        self.negative_cache_remove(&new_full_path);
+        // Cancel any queued remote delete for the destination path (rm b && ln a b).
+        if let Some(fm) = &self.flush_manager {
+            fm.cancel_delete(&new_full_path);
+        }
+
+        // Phase 3: insert the alias into the inode table under the write lock.
+        let mut inodes = self.inode_table.write().expect("inodes poisoned");
+        if inodes.get(newparent).is_none() {
+            return Err(libc::ENOENT);
+        }
+        // A concurrent create may have won the race since phase 1; the remote
+        // add already landed, but locally the newer entry owns the name.
+        if inodes.lookup_child(newparent, newname).is_some() {
+            return Err(libc::EEXIST);
+        }
+        info!("Linked {} -> {} (server-side copy)", source_path, new_full_path);
+        let new_ino = inodes.insert(
+            newparent,
+            newname.to_string(),
+            new_full_path,
+            InodeKind::File,
+            size,
+            now,
+            Some(xet_hash),
+            mode,
+            uid,
+            gid,
+        );
+        inodes.touch_parent(newparent, now);
+        inodes.touch(new_ino);
+
+        Ok(self.make_vfs_attr(inodes.get(new_ino).ok_or(libc::EIO)?))
+    }
+
     pub async fn unlink(&self, parent: u64, name: &str) -> VirtualFsResult<()> {
         if self.read_only {
             return Err(libc::EROFS);
@@ -3091,12 +3215,6 @@ impl VirtualFs {
         }
     }
 
-    pub async fn link(&self, _ino: u64, _new_parent: u64, _new_name: &str) -> VirtualFsResult<VirtualFsAttr> {
-        // Hard links are not supported — they are ephemeral (in-memory only) and never
-        // persisted to the hub, which makes them a source of subtle bugs with no benefit.
-        Err(libc::ENOTSUP)
-    }
-
     pub async fn rmdir(&self, parent: u64, name: &str) -> VirtualFsResult<()> {
         if self.read_only {
             return Err(libc::EROFS);

src/virtual_fs/tests.rs

@@ -4458,6 +4458,208 @@ fn rename_clean_file_remote_and_local() {
     });
 }
 
+// ── link(): server-side copy ─────────────────────────────────────────────
+
+/// link() on a clean remote file commits one AddFile with the source's xet
+/// hash (no data upload) and inserts a separate alias inode locally.
+#[test]
+fn link_clean_file_creates_server_side_copy() {
+    let hub = MockHub::new();
+    hub.add_file("src.txt", 100, Some("hash1"), None);
+    let xet = MockXet::new();
+    let (rt, vfs) = vfs_simple(&hub, &xet);
+
+    rt.block_on(async {
+        let src_attr = vfs.lookup(ROOT_INODE, "src.txt").await.unwrap();
+
+        let alias_attr = vfs.link(src_attr.ino, ROOT_INODE, "alias.txt").await.unwrap();
+        assert_ne!(alias_attr.ino, src_attr.ino, "alias gets its own inode");
+        assert_eq!(alias_attr.size, 100);
+
+        let logs = hub.take_batch_log();
+        assert_eq!(logs.len(), 1, "exactly one batch commit");
+        assert_eq!(logs[0].len(), 1, "a single AddFile, no delete");
+        match &logs[0][0] {
+            BatchOp::AddFile { path, xet_hash, .. } => {
+                assert_eq!(path, "alias.txt");
+                assert_eq!(xet_hash, "hash1");
+            }
+            other => panic!("expected AddFile, got {other:?}"),
+        }
+
+        // The alias is a real entry: resolvable, right hash, source untouched.
+        let inodes = vfs.inode_table.read().unwrap();
+        let alias = inodes.get(alias_attr.ino).unwrap();
+        assert_eq!(alias.full_path.as_ref(), "alias.txt");
+        assert_eq!(alias.xet_hash.as_deref(), Some("hash1"));
+        assert!(!alias.is_dirty());
+        let src = inodes.get(src_attr.ino).unwrap();
+        assert_eq!(src.full_path.as_ref(), "src.txt");
+    });
+}
+
+/// link() on a dirty source must not alias a stale hash: ENOTSUP so the
+/// caller falls back to a regular copy.
+#[test]
+fn link_dirty_source_not_supported() {
+    let hub = MockHub::new();
+    hub.add_file("src.txt", 100, Some("hash1"), None);
+    let xet = MockXet::new();
+    let (rt, vfs) = vfs_simple(&hub, &xet);
+
+    rt.block_on(async {
+        let src_attr = vfs.lookup(ROOT_INODE, "src.txt").await.unwrap();
+        {
+            let mut inodes = vfs.inode_table.write().unwrap();
+            inodes.get_mut(src_attr.ino).unwrap().set_dirty();
+        }
+
+        let err = vfs.link(src_attr.ino, ROOT_INODE, "alias.txt").await.unwrap_err();
+        assert_eq!(err, libc::ENOTSUP);
+        assert!(hub.take_batch_log().is_empty(), "no remote op for a rejected link");
+    });
+}
+
+/// link() onto an existing name is EEXIST and sends no remote op.
+#[test]
+fn link_existing_target_eexist() {
+    let hub = MockHub::new();
+    hub.add_file("src.txt", 100, Some("hash1"), None);
+    hub.add_file("dst.txt", 50, Some("hash2"), None);
+    let xet = MockXet::new();
+    let (rt, vfs) = vfs_simple(&hub, &xet);
+
+    rt.block_on(async {
+        let src_attr = vfs.lookup(ROOT_INODE, "src.txt").await.unwrap();
+        let err = vfs.link(src_attr.ino, ROOT_INODE, "dst.txt").await.unwrap_err();
+        assert_eq!(err, libc::EEXIST);
+        assert!(hub.take_batch_log().is_empty());
+    });
+}
+
+/// link() on a directory is EPERM (POSIX).
+#[test]
+fn link_directory_eperm() {
+    let hub = MockHub::new();
+    hub.add_file("d/x.txt", 10, Some("hash1"), None);
+    let xet = MockXet::new();
+    let (rt, vfs) = vfs_simple(&hub, &xet);
+
+    rt.block_on(async {
+        let dir_attr = vfs.lookup(ROOT_INODE, "d").await.unwrap();
+        let err = vfs.link(dir_attr.ino, ROOT_INODE, "d2").await.unwrap_err();
+        assert_eq!(err, libc::EPERM);
+        assert!(hub.take_batch_log().is_empty());
+    });
+}
+
+/// link() on a clean file whose committed hash is missing/empty returns
+/// ENOTSUP (we never alias a hashless entry), with no remote op.
+#[test]
+fn link_no_committed_hash_not_supported() {
+    let hub = MockHub::new();
+    hub.add_file("src.txt", 100, Some("hash1"), None);
+    let xet = MockXet::new();
+    let (rt, vfs) = vfs_simple(&hub, &xet);
+
+    rt.block_on(async {
+        let src_attr = vfs.lookup(ROOT_INODE, "src.txt").await.unwrap();
+        {
+            let mut inodes = vfs.inode_table.write().unwrap();
+            inodes.get_mut(src_attr.ino).unwrap().xet_hash = None;
+        }
+
+        let err = vfs.link(src_attr.ino, ROOT_INODE, "alias.txt").await.unwrap_err();
+        assert_eq!(err, libc::ENOTSUP);
+        assert!(hub.take_batch_log().is_empty(), "no remote op for a rejected link");
+    });
+}
+
+/// link() into a missing parent directory is ENOENT, with no remote op.
+#[test]
+fn link_missing_parent_enoent() {
+    let hub = MockHub::new();
+    hub.add_file("src.txt", 100, Some("hash1"), None);
+    let xet = MockXet::new();
+    let (rt, vfs) = vfs_simple(&hub, &xet);
+
+    rt.block_on(async {
+        let src_attr = vfs.lookup(ROOT_INODE, "src.txt").await.unwrap();
+        let err = vfs.link(src_attr.ino, 999_999, "alias.txt").await.unwrap_err();
+        assert_eq!(err, libc::ENOENT);
+        assert!(hub.take_batch_log().is_empty());
+    });
+}
+
+/// link() into a non-directory parent is ENOTDIR, with no remote op.
+#[test]
+fn link_nondir_parent_enotdir() {
+    let hub = MockHub::new();
+    hub.add_file("src.txt", 100, Some("hash1"), None);
+    hub.add_file("file.txt", 50, Some("hash2"), None);
+    let xet = MockXet::new();
+    let (rt, vfs) = vfs_simple(&hub, &xet);
+
+    rt.block_on(async {
+        let src_attr = vfs.lookup(ROOT_INODE, "src.txt").await.unwrap();
+        let file_attr = vfs.lookup(ROOT_INODE, "file.txt").await.unwrap();
+        let err = vfs.link(src_attr.ino, file_attr.ino, "alias.txt").await.unwrap_err();
+        assert_eq!(err, libc::ENOTDIR);
+        assert!(hub.take_batch_log().is_empty());
+    });
+}
+
+/// link() to an OS junk name is rejected with EACCES, with no remote op.
+#[test]
+fn link_os_junk_name_eacces() {
+    let hub = MockHub::new();
+    hub.add_file("src.txt", 100, Some("hash1"), None);
+    let xet = MockXet::new();
+    let (rt, vfs) = vfs_simple(&hub, &xet);
+
+    rt.block_on(async {
+        let src_attr = vfs.lookup(ROOT_INODE, "src.txt").await.unwrap();
+        let err = vfs.link(src_attr.ino, ROOT_INODE, ".DS_Store").await.unwrap_err();
+        assert_eq!(err, libc::EACCES);
+        assert!(hub.take_batch_log().is_empty());
+    });
+}
+
+/// link() on a read-only mount is EROFS, with no remote op.
+#[test]
+fn link_read_only_erofs() {
+    let hub = MockHub::new();
+    hub.add_file("src.txt", 100, Some("hash1"), None);
+    let xet = MockXet::new();
+    let (rt, vfs) = vfs_readonly(&hub, &xet);
+
+    rt.block_on(async {
+        let src_attr = vfs.lookup(ROOT_INODE, "src.txt").await.unwrap();
+        let err = vfs.link(src_attr.ino, ROOT_INODE, "alias.txt").await.unwrap_err();
+        assert_eq!(err, libc::EROFS);
+        assert!(hub.take_batch_log().is_empty());
+    });
+}
+
+/// link() in overlay mode is ENOTSUP (committing an alias would mutate the
+/// remote behind the overlay's back), with no remote op.
+#[test]
+fn link_overlay_not_supported() {
+    let hub = MockHub::new();
+    hub.add_file("src.txt", 100, Some("hash1"), None);
+    let xet = MockXet::new();
+
+    let overlay_root = fresh_overlay_dir("link");
+    let t = make_overlay_test_vfs_with_root(hub.clone(), xet, overlay_root);
+
+    t.runtime.block_on(async {
+        let src_attr = t.vfs.lookup(ROOT_INODE, "src.txt").await.unwrap();
+        let err = t.vfs.link(src_attr.ino, ROOT_INODE, "alias.txt").await.unwrap_err();
+        assert_eq!(err, libc::ENOTSUP);
+        assert!(hub.take_batch_log().is_empty());
+    });
+}
+
 // ── LRU eviction: end-to-end memory bound ───────────────────────────────
 
 /// End-to-end test of the forget-driven eviction path on a hierarchical