corrected large file issue

Author: ehsan6sha

crates/fula-blockstore/src/ipfs.rs

@@ -158,11 +158,15 @@ impl IpfsBlockStore {
     }
 
     /// Put a raw block
+    /// Note: Raw blocks (file chunks) are NOT pinned inline to avoid timeouts.
+    /// They are protected by the bucket root's recursive pin after flush_forest.
     #[instrument(skip(self, data), fields(size = data.len()))]
     pub async fn put_block_raw(&self, data: &[u8]) -> Result<Cid> {
+        // Don't pin raw blocks inline - even small blocks can cause timeouts under load.
+        // All blocks are protected by recursive pinning of the bucket root CID at flush time.
         let url = format!(
             "{}/api/v0/block/put?cid-codec=raw&mhtype=blake3",
-            self.config.api_url
+            self.config.api_url,
         );
 
         let part = multipart::Part::bytes(data.to_vec())
@@ -267,6 +271,8 @@ impl BlockStore for IpfsBlockStore {
 
         tracing::debug!(bytes_len = bytes.len(), "Serialized IPLD data to CBOR");
 
+        // Don't pin inline - it delays the response body and causes timeouts.
+        // All data is protected by bucket root recursive pin at flush_forest time.
         let url = format!(
             "{}/api/v0/dag/put?store-codec=dag-cbor&input-codec=dag-cbor",
             self.config.api_url
@@ -294,10 +300,14 @@ impl BlockStore for IpfsBlockStore {
             )));
         }
 
-        let result: DagPutResponse = response
-            .json()
-            .await
-            .map_err(|e| BlockStoreError::IpfsApi(e.to_string()))?;
+        // Read response body as text first for debugging
+        let body_text = response.text().await
+            .map_err(|e| BlockStoreError::IpfsApi(format!("Failed to read response body: {}", e)))?;
+        
+        tracing::debug!(body = %body_text, "IPFS dag/put response body");
+        
+        let result: DagPutResponse = serde_json::from_str(&body_text)
+            .map_err(|e| BlockStoreError::IpfsApi(format!("Failed to parse response: {} - body: {}", e, body_text)))?;
 
         tracing::debug!(cid = %result.cid.root_cid, "DAG stored successfully");
 

crates/fula-blockstore/src/ipfs_pinning.rs

@@ -242,10 +242,14 @@ impl BlockStore for IpfsPinningBlockStore {
             self.cache.insert(cid, Bytes::copy_from_slice(data));
         }
 
-        // Pin for persistence
-        if let Err(e) = self.pin_cid(&cid, None).await {
-            warn!(cid = %cid, error = %e, "Failed to pin CID, data may not be persistent");
-        }
+        // NOTE: We deliberately DO NOT pin individual blocks here.
+        // Instead, the root CID should be pinned at upload completion.
+        // IPFS recursive pinning will automatically pin all referenced blocks.
+        // This optimization reduces N pin requests to 1 for chunked uploads.
+        //
+        // Pinning should happen at:
+        // - put_object() completion for single objects
+        // - complete_multipart_upload() for multipart uploads
 
         Ok(cid)
     }
@@ -302,10 +306,8 @@ impl BlockStore for IpfsPinningBlockStore {
     async fn put_ipld<T: serde::Serialize + Send + Sync>(&self, data: &T) -> Result<Cid> {
         let cid = self.ipfs.put_ipld(data).await?;
 
-        // Pin for persistence
-        if let Err(e) = self.pin_cid(&cid, None).await {
-            warn!(cid = %cid, error = %e, "Failed to pin IPLD, data may not be persistent");
-        }
+        // NOTE: We deliberately DO NOT pin IPLD nodes inline.
+        // Pin the root CID at upload completion for recursive pinning.
 
         Ok(cid)
     }
@@ -491,6 +493,47 @@ impl BlockStore for FlexibleBlockStore {
     }
 }
 
+#[async_trait]
+impl PinStore for FlexibleBlockStore {
+    async fn pin(&self, cid: &Cid, name: Option<&str>) -> Result<()> {
+        match self {
+            Self::IpfsPinning(store) => store.pin(cid, name).await,
+            Self::Memory(_) => {
+                // Memory store doesn't need pinning, just succeed silently
+                Ok(())
+            }
+        }
+    }
+
+    async fn unpin(&self, cid: &Cid) -> Result<()> {
+        match self {
+            Self::IpfsPinning(store) => store.unpin(cid).await,
+            Self::Memory(_) => Ok(()),
+        }
+    }
+
+    async fn is_pinned(&self, cid: &Cid) -> Result<bool> {
+        match self {
+            Self::IpfsPinning(store) => store.is_pinned(cid).await,
+            Self::Memory(_) => Ok(true), // Memory store "pins" everything
+        }
+    }
+
+    async fn list_pins(&self) -> Result<Vec<Cid>> {
+        match self {
+            Self::IpfsPinning(store) => store.list_pins().await,
+            Self::Memory(_) => Ok(Vec::new()),
+        }
+    }
+
+    async fn pin_status(&self, cid: &Cid) -> Result<crate::PinStatus> {
+        match self {
+            Self::IpfsPinning(store) => store.pin_status(cid).await,
+            Self::Memory(_) => Ok(crate::PinStatus::Pinned), // Memory store "pins" everything
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/fula-cli/src/error.rs

@@ -189,6 +189,14 @@ impl IntoResponse for ApiError {
             ApiError::S3Error { request_id, .. } => request_id.clone(),
             _ => uuid::Uuid::new_v4().to_string(),
         };
+        
+        // Log the actual error for debugging
+        tracing::error!(
+            error_code = %code.as_str(),
+            status = %status,
+            error = %self,
+            "API error response"
+        );
 
         let xml = format!(
             r#"<?xml version="1.0" encoding="UTF-8"?>

crates/fula-cli/src/handlers/multipart.rs

@@ -1,6 +1,7 @@
 //! Multipart upload handlers
 
 use crate::{AppState, ApiError, S3ErrorCode};
+use crate::pinning::pin_for_user;
 use crate::state::UserSession;
 use crate::multipart::UploadPart;
 use crate::xml;
@@ -10,7 +11,7 @@ use axum::{
     response::{IntoResponse, Response},
 };
 use bytes::Bytes;
-use fula_blockstore::BlockStore;
+use fula_blockstore::{BlockStore, PinStore};
 use fula_core::metadata::ObjectMetadata;
 use fula_crypto::hashing::md5_hash;
 use serde::Deserialize;
@@ -139,6 +140,7 @@ pub async fn complete_multipart_upload(
     Extension(session): Extension<UserSession>,
     Path((bucket, key)): Path<(String, String)>,
     Query(params): Query<MultipartParams>,
+    headers: HeaderMap,
     _body: Bytes,
 ) -> Result<Response, ApiError> {
     if !session.can_write() {
@@ -167,10 +169,15 @@ pub async fn complete_multipart_upload(
     // Calculate total size
     let total_size: u64 = upload.parts.values().map(|p| p.size).sum();
 
+    // Collect all part CIDs for pinning
+    let part_cids: Vec<cid::Cid> = upload.parts.values()
+        .filter_map(|p| p.cid.parse().ok())
+        .collect();
+
     // Create the final object metadata
     // In a real implementation, we'd create a DAG linking all parts
-    let first_part_cid: cid::Cid = upload.parts.values().next()
-        .map(|p| p.cid.parse().unwrap())
+    let first_part_cid: cid::Cid = part_cids.first()
+        .copied()
         .ok_or_else(|| ApiError::s3(S3ErrorCode::InvalidPart, "No parts uploaded"))?;
 
     let mut metadata = ObjectMetadata::new(first_part_cid, total_size, final_etag.clone())
@@ -187,7 +194,27 @@ pub async fn complete_multipart_upload(
     // Store in bucket
     let mut bucket_handle = state.bucket_manager.open_bucket(&bucket).await?;
     bucket_handle.put_object(key.clone(), metadata).await?;
-    bucket_handle.flush().await?;
+    let bucket_root_cid = bucket_handle.flush().await?;
+
+    // Pin the BUCKET ROOT CID to ensure tree structure survives GC.
+    // This recursively pins all tree nodes AND all referenced object data (including parts).
+    // NOTE: Pinning is async (fire-and-forget) to avoid blocking the response.
+    {
+        let block_store = Arc::clone(&state.block_store);
+        let pin_bucket = bucket.clone();
+        tokio::spawn(async move {
+            let pin_name = format!("bucket:{}", pin_bucket);
+            if let Err(e) = block_store.pin(&bucket_root_cid, Some(&pin_name)).await {
+                tracing::warn!(cid = %bucket_root_cid, error = %e, "Failed to pin bucket root CID");
+            } else {
+                tracing::info!(cid = %bucket_root_cid, bucket = %pin_name, "Bucket root CID pinned (recursive)");
+            }
+        });
+    }
+
+    // Also pin to user's external pinning service if credentials provided
+    // Pin the first part CID as the representative (or all parts)
+    pin_for_user(&headers, &first_part_cid, Some(&key)).await;
 
     let location = format!("/{}/{}", bucket, key);
     let xml_response = xml::complete_multipart_upload_result(

crates/fula-cli/src/handlers/object.rs

@@ -11,7 +11,7 @@ use axum::{
     response::{IntoResponse, Response},
 };
 use bytes::Bytes;
-use fula_blockstore::BlockStore;
+use fula_blockstore::{BlockStore, PinStore};
 use fula_core::metadata::ObjectMetadata;
 use fula_crypto::hashing::md5_hash;
 use serde::Deserialize;
@@ -91,13 +91,28 @@ pub async fn put_object(
         })?;
 
     tracing::debug!("Flushing bucket");
-    bucket.flush().await
+    let bucket_root_cid = bucket.flush().await
         .map_err(|e| {
             tracing::error!(error = %e, "Failed to flush bucket");
             e
         })?;
 
-    // Pin to user's pinning service if credentials provided
+    // Pin the BUCKET ROOT CID to ensure tree structure survives GC.
+    // This recursively pins all tree nodes AND all referenced object data.
+    // NOTE: Pinning is async (fire-and-forget) to avoid blocking the response.
+    {
+        let block_store = Arc::clone(&state.block_store);
+        let pin_name = format!("bucket:{}", bucket_name);
+        tokio::spawn(async move {
+            if let Err(e) = block_store.pin(&bucket_root_cid, Some(&pin_name)).await {
+                tracing::warn!(cid = %bucket_root_cid, error = %e, "Failed to pin bucket root CID");
+            } else {
+                tracing::info!(cid = %bucket_root_cid, bucket = %pin_name, "Bucket root CID pinned (recursive)");
+            }
+        });
+    }
+
+    // Also pin to user's external pinning service if credentials provided
     // Headers: X-Pinning-Service, X-Pinning-Token
     pin_for_user(&headers, &cid, Some(&key)).await;
 

crates/fula-cli/src/routes.rs

@@ -220,7 +220,7 @@ async fn object_post_handler(
     if query.uploads.is_some() {
         handlers::create_multipart_upload(state, session, path, headers).await
     } else if query.upload_id.is_some() {
-        handlers::complete_multipart_upload(state, session, path, query, body).await
+        handlers::complete_multipart_upload(state, session, path, query, headers, body).await
     } else {
         Err(crate::ApiError::s3(
             crate::S3ErrorCode::InvalidRequest,