fix(vfs): surface real Hub error and map HTTP status to errno on write failure (#198)

## Problem

When a streaming (append-only) write fails, two diagnosability problems
stack up:

1. The streaming worker captures the real error from `writer.write()`,
but the `Finish` handler throws it away and returns a hardcoded
`Error::hub("streaming write failed")`. The log only shows a generic
message with no Hub status.
2. Every commit failure maps to a blanket `EIO`, so an importing client
(e.g. Radarr/Sonarr) cannot tell a quota/storage reject apart from a
real I/O error, and retries straight into the quota wall.

This surfaced during a real quota incident: the Hub rejected writes
(quota exceeded), but the log only said `Hub API error: streaming write
failed` with `errno=5`. The actual cause had to be guessed by hand, and
the importing app kept looping with file loss.

## Changes

- Keep the structured `Error` (not just its string) in the streaming
worker so the HTTP status survives, and propagate it on `Finish`.
- Preserve the HTTP status from CAS `ClientError` in
`From<xet_data::DataError>` instead of flattening it into an opaque
string. (Verified the upload error path keeps the structured
`DataError::ClientError`, whose `status()` exposes the code.)
- Add `Error::status()` and `Error::to_errno()`: map `413`/`507` to
`ENOSPC`, `403` to `EACCES`, `429` to `EAGAIN`, everything else stays
`EIO`.
- Return `e.to_errno()` from the streaming write/commit failure paths.

## Result

A quota reject now:
- logs the real `Hub API error (413): ...` instead of a generic string,
and
- surfaces as `ENOSPC` ("No space left on device") to the FUSE client,
so the importing app stops retrying into a quota wall instead of
looping.

## Design notes

- `413`/`507` map to `ENOSPC` rather than `EDQUOT`: `ENOSPC` is more
widely recognized by importing apps and clearer to operators. Happy to
switch to `EDQUOT` if preferred.
- Scope is limited to the streaming/append-only write path (the incident
path). The advanced-writes flush path (which already retries) and remote
deletes are untouched.

## Tests

- `error::tests` covers the status -> errno mapping (including
statusless and non-Hub errors staying `EIO`).
- `streaming_worker_surfaces_real_hub_error_on_failed_write` is a
regression test: a write failing with a `413` must surface the real
message, `status() == Some(413)`, and `to_errno() == ENOSPC`.
- Full suite green with `--features fuse,nfs` (349 passed); clippy clean
across all feature combos with `-D warnings`.

Author: XciD

src/error.rs

@@ -23,6 +23,29 @@ impl Error {
             status: Some(status),
         }
     }
+
+    /// HTTP status carried by the error, when it originated from a Hub/CAS response.
+    pub fn status(&self) -> Option<u16> {
+        match self {
+            Self::Hub { status, .. } => *status,
+            _ => None,
+        }
+    }
+
+    /// Errno to surface to FUSE clients. Maps known Hub/CAS HTTP statuses to a
+    /// meaningful errno so an importing app (e.g. Radarr/Sonarr) can tell a quota
+    /// or storage reject apart from a generic I/O failure. Everything else stays EIO.
+    pub fn to_errno(&self) -> i32 {
+        match self.status() {
+            // Payload too large / insufficient storage: surface as "no space left"
+            // so the client stops retrying into a quota wall instead of looping.
+            Some(413 | 507) => libc::ENOSPC,
+            Some(403) => libc::EACCES,
+            // Rate-limited: transient, signal "try again".
+            Some(429) => libc::EAGAIN,
+            _ => libc::EIO,
+        }
+    }
 }
 
 impl fmt::Display for Error {
@@ -63,6 +86,14 @@ impl From<reqwest::Error> for Error {
 
 impl From<xet_data::DataError> for Error {
     fn from(err: xet_data::DataError) -> Self {
+        // Preserve the HTTP status from CAS client errors (e.g. 413/507 on a quota
+        // reject during upload) so it can be surfaced in logs and mapped to a
+        // meaningful errno. Without this the status is flattened into an opaque string.
+        if let xet_data::DataError::ClientError(client_error) = &err
+            && let Some(status) = client_error.status()
+        {
+            return Self::hub_status(status.as_u16(), err.to_string());
+        }
         Self::Xet(err.to_string())
     }
 }
@@ -78,3 +109,27 @@ pub fn is_retryable_status(status: u16) -> bool {
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn to_errno_maps_known_statuses() {
+        assert_eq!(Error::hub_status(413, "payload too large").to_errno(), libc::ENOSPC);
+        assert_eq!(Error::hub_status(507, "insufficient storage").to_errno(), libc::ENOSPC);
+        assert_eq!(Error::hub_status(403, "forbidden").to_errno(), libc::EACCES);
+        assert_eq!(Error::hub_status(429, "rate limited").to_errno(), libc::EAGAIN);
+        // Unknown status and statusless errors stay generic.
+        assert_eq!(Error::hub_status(500, "server error").to_errno(), libc::EIO);
+        assert_eq!(Error::hub("no status").to_errno(), libc::EIO);
+        assert_eq!(Error::Xet("opaque".into()).to_errno(), libc::EIO);
+    }
+
+    #[test]
+    fn status_only_set_for_hub_with_status() {
+        assert_eq!(Error::hub_status(413, "x").status(), Some(413));
+        assert_eq!(Error::hub("x").status(), None);
+        assert_eq!(Error::Xet("x".into()).status(), None);
+    }
+}

src/virtual_fs/mod.rs

@@ -1120,7 +1120,7 @@ impl VirtualFs {
         // unbounded memory. 32 slots × ~128KB FUSE write = ~4MB max in-flight.
         // blocking_send is safe here: FUSE threads are not tokio workers.
         let (tx, rx) = tokio::sync::mpsc::channel::<WriteMsg>(32);
-        let error: Arc<std::sync::Mutex<Option<String>>> = Arc::new(std::sync::Mutex::new(None));
+        let error: Arc<std::sync::Mutex<Option<crate::error::Error>>> = Arc::new(std::sync::Mutex::new(None));
         self.runtime
             .spawn(streaming_worker(streaming_writer, rx, error.clone()));
 
@@ -2316,8 +2316,8 @@ impl VirtualFs {
                 channel,
             } => {
                 // Check for previous worker error
-                if channel.error.lock().expect("error poisoned").is_some() {
-                    return Err(libc::EIO);
+                if let Some(err) = channel.error.lock().expect("error poisoned").as_ref() {
+                    return Err(err.to_errno());
                 }
 
                 // Enforce append-only: offset must match bytes written so far
@@ -2362,8 +2362,8 @@ impl VirtualFs {
 
         if let Some(channel) = streaming_channel {
             // Check for worker errors first.
-            if channel.error.lock().expect("error poisoned").is_some() {
-                return Err(libc::EIO);
+            if let Some(err) = channel.error.lock().expect("error poisoned").as_ref() {
+                return Err(err.to_errno());
             }
 
             // Check current commit state.
@@ -2629,7 +2629,7 @@ impl VirtualFs {
                     Ok(Ok(info)) => info,
                     Ok(Err(e)) => {
                         error!("Streaming upload failed for ino={}: {}", ino, e);
-                        return Err(libc::EIO);
+                        return Err(e.to_errno());
                     }
                     Err(_) => {
                         error!("Streaming worker dropped for ino={}", ino);
@@ -2665,7 +2665,7 @@ impl VirtualFs {
             error!("Failed to commit file {}: {}", full_path, e);
             // CAS upload succeeded — preserve file_info for retry in release()
             *channel.pending_info.lock().expect("pending_info poisoned") = Some(file_info);
-            return Err(libc::EIO);
+            return Err(e.to_errno());
         }
 
         let mut inodes = self.inode_table.write().expect("inodes poisoned");
@@ -3837,7 +3837,8 @@ struct StreamingChannel {
     tx: tokio::sync::mpsc::Sender<WriteMsg>,
     bytes_written: AtomicU64,
     /// Set by the background worker if add_data() fails. Shared with worker via Arc.
-    error: Arc<std::sync::Mutex<Option<String>>>,
+    /// Holds the structured error so its HTTP status survives to errno mapping.
+    error: Arc<std::sync::Mutex<Option<crate::error::Error>>>,
     /// Commit lifecycle state machine.
     state: std::sync::Mutex<CommitState>,
     /// CAS upload succeeded but Hub commit failed — stored for retry.
@@ -3904,7 +3905,7 @@ fn same_process(pid_a: u32, pid_b: u32) -> bool {
 async fn streaming_worker(
     mut writer: Box<dyn StreamingWriterOps>,
     mut rx: tokio::sync::mpsc::Receiver<WriteMsg>,
-    error: Arc<std::sync::Mutex<Option<String>>>,
+    error: Arc<std::sync::Mutex<Option<crate::error::Error>>>,
 ) {
     let mut failed = false;
     while let Some(msg) = rx.recv().await {
@@ -3914,13 +3915,21 @@ async fn streaming_worker(
                     continue; // drain remaining messages
                 }
                 if let Err(e) = writer.write(&data).await {
-                    *error.lock().unwrap() = Some(e.to_string());
+                    *error.lock().expect("error poisoned") = Some(e);
                     failed = true;
                 }
             }
             WriteMsg::Finish(reply) => {
                 let result = if failed {
-                    Err(crate::error::Error::hub("streaming write failed"))
+                    // Surface the real error captured on the failing write() instead of a
+                    // hardcoded generic. Keeping the structured Error preserves its HTTP
+                    // status (e.g. 413/507 on a quota reject), which both logs the real
+                    // cause and lets streaming_commit map it to a meaningful errno.
+                    Err(error
+                        .lock()
+                        .expect("error poisoned")
+                        .take()
+                        .unwrap_or_else(|| crate::error::Error::hub("streaming write failed: unknown error")))
                 } else {
                     writer.finish_boxed().await
                 };

src/virtual_fs/tests.rs

@@ -5418,3 +5418,50 @@ fn overlay_rmdir_remote_dir_eperm() {
         assert_eq!(err, libc::EPERM);
     });
 }
+
+/// Regression: when a streaming write fails, the worker must surface the real Hub
+/// error (including its HTTP status) on Finish, not a hardcoded generic. A quota
+/// reject (HTTP 413) was previously logged as "Hub API error: streaming write failed"
+/// and mapped to a blanket EIO, hiding the actual cause and making the importing app
+/// (Radarr/Sonarr) retry into a quota wall instead of stopping.
+#[test]
+fn streaming_worker_surfaces_real_hub_error_on_failed_write() {
+    struct FailingWriter;
+
+    #[async_trait::async_trait]
+    impl StreamingWriterOps for FailingWriter {
+        async fn write(&mut self, _data: &[u8]) -> crate::error::Result<()> {
+            Err(crate::error::Error::hub_status(413, "storage quota exceeded"))
+        }
+        async fn finish_boxed(self: Box<Self>) -> crate::error::Result<XetFileInfo> {
+            unreachable!("finish must not be called after a failed write");
+        }
+        fn len(&self) -> u64 {
+            0
+        }
+        fn is_empty(&self) -> bool {
+            true
+        }
+    }
+
+    new_runtime().block_on(async {
+        let (tx, rx) = tokio::sync::mpsc::channel::<WriteMsg>(4);
+        let error = Arc::new(std::sync::Mutex::new(None));
+        let worker = tokio::spawn(streaming_worker(Box::new(FailingWriter), rx, error.clone()));
+
+        tx.send(WriteMsg::Data(vec![0u8; 8])).await.unwrap();
+        let (result_tx, result_rx) = tokio::sync::oneshot::channel();
+        tx.send(WriteMsg::Finish(result_tx)).await.unwrap();
+
+        let result = result_rx.await.expect("worker dropped reply");
+        let err = result.expect_err("failed write must produce an error");
+        let msg = err.to_string();
+        assert!(msg.contains("413"), "lost HTTP status: {msg}");
+        assert!(msg.contains("storage quota exceeded"), "lost real detail: {msg}");
+        // Structured status must survive so it maps to a meaningful errno, not blanket EIO.
+        assert_eq!(err.status(), Some(413));
+        assert_eq!(err.to_errno(), libc::ENOSPC);
+
+        worker.await.unwrap();
+    });
+}