fix: Analyze and simply handle consistency issues between the database and S3.

ceres/src/api_service/admin_ops.rs

@@ -94,11 +94,23 @@ impl MonoApiService {
                 MegaError::Other(format!("{} not found in root directory", ADMIN_FILE))
             })?;
 
-        let content_bytes = self
+        let blob_hash = blob_item.id.to_string();
+        let content_bytes = match self
             .storage
             .git_service
-            .get_object_as_bytes(&blob_item.id.to_string())
-            .await?;
+            .get_object_as_bytes(&blob_hash)
+            .await
+        {
+            Ok(bytes) => bytes,
+            Err(e) => {
+                // Best-effort classification/logging for ObjStorageNotFound cases.
+                let e = self
+                    .storage
+                    .classify_blob_objstorage_not_found(&blob_hash, e)
+                    .await;
+                return Err(e);
+            }
+        };
 
         let content = String::from_utf8(content_bytes)
             .map_err(|e| MegaError::Other(format!("UTF-8 decode failed: {}", e)))?;

ceres/src/api_service/commit_ops.rs

@@ -708,10 +708,12 @@ async fn compute_commit_diff_items<T: ApiHandler + ?Sized>(
         parent_blobs_set.push(blobs);
     }
 
-    let ctx = handler.get_context();
     let mut blob_cache: HashMap<ObjectHash, Vec<u8>> = HashMap::new();
     for hash in &all_hashes {
-        match ctx.git_service.get_object_as_bytes(&hash.to_string()).await {
+        // Reuse ApiHandler's blob reading helper so that:
+        // - object IDs are validated
+        // - ObjStorageNotFound cases are classified/logged via Storage helpers
+        match handler.get_raw_blob_by_hash(&hash.to_string()).await {
             Ok(blob) => {
                 blob_cache.insert(*hash, blob);
             }

ceres/src/api_service/mod.rs

@@ -122,10 +122,15 @@ pub trait ApiHandler: Send + Sync {
     ) -> Result<CreateEntryResult, GitError>;
 
     async fn get_raw_blob_by_hash(&self, hash: &str) -> Result<Vec<u8>, MegaError> {
-        self.get_context()
-            .git_service
-            .get_object_as_bytes(hash)
-            .await
+        let storage = self.get_context();
+        match storage.git_service.get_object_as_bytes(hash).await {
+            Ok(data) => Ok(data),
+            Err(e) => {
+                // Best-effort classification/logging for ObjStorageNotFound cases.
+                let e = storage.classify_blob_objstorage_not_found(hash, e).await;
+                Err(e)
+            }
+        }
     }
 
     /// Preview unified diff for a single file change

ceres/src/api_service/mono_api_service.rs

@@ -636,23 +636,23 @@ impl ApiHandler for MonoApiService {
     }
 
     async fn get_tree_by_hash(&self, hash: &str) -> Result<Tree, MegaError> {
-        Ok(Tree::from_mega_model(
-            self.storage
-                .mono_storage()
-                .get_tree_by_hash(hash)
-                .await?
-                .unwrap_or_else(|| panic!("can't fetch tree by hash {}", hash)),
-        ))
+        let model = self
+            .storage
+            .mono_storage()
+            .get_tree_by_hash(hash)
+            .await?
+            .ok_or_else(|| MegaError::NotFound(format!("tree not found: {}", hash)))?;
+        Ok(Tree::from_mega_model(model))
     }
 
     async fn get_commit_by_hash(&self, hash: &str) -> Result<Commit, MegaError> {
-        Ok(Commit::from_mega_model(
-            self.storage
-                .mono_storage()
-                .get_commit_by_hash(hash)
-                .await?
-                .unwrap_or_else(|| panic!("can't fetch commit by hash {}", hash)),
-        ))
+        let model = self
+            .storage
+            .mono_storage()
+            .get_commit_by_hash(hash)
+            .await?
+            .ok_or_else(|| MegaError::NotFound(format!("commit not found: {}", hash)))?;
+        Ok(Commit::from_mega_model(model))
     }
 
     async fn item_to_commit_map(

ceres/src/lfs/handler.rs

@@ -522,7 +522,7 @@ async fn lfs_download_url(
 
     if let Some(url) = storage
         .inner
-        .signed_url(&key, Method::GET, Duration::from_hours(1))
+        .signed_url(&key, Method::GET, Duration::from_secs(3600))
         .await?
     {
         return Ok(url);
@@ -540,7 +540,7 @@ async fn lfs_upload_url(
 
     if let Some(url) = storage
         .inner
-        .signed_url(&key, Method::PUT, Duration::from_hours(1))
+        .signed_url(&key, Method::PUT, Duration::from_secs(3600))
         .await?
     {
         return Ok(url);

common/src/errors.rs

@@ -64,6 +64,15 @@ pub enum MegaError {
     #[error("ObjStorage error: {0}")]
     ObjStorage(String),
 
+    /// Object not found in underlying object storage (S3/GCS/local).
+    /// Typically corresponds to a 404/NoSuchKey-style error.
+    #[error("ObjStorage not found: {0}")]
+    ObjStorageNotFound(String),
+
+    ///Object not found in underlying object storage (S3/GCS/local). but exists in the repository.
+    #[error("ObjStorage inconsistent: {0}")]
+    ObjStorageInconsistent(String),
+
     // --- Other ---
     #[error("Other error: {0}")]
     Other(String),
@@ -83,7 +92,14 @@ impl From<ParseErrors> for MegaError {
 
 impl From<MegaError> for GitError {
     fn from(val: MegaError) -> Self {
-        GitError::CustomError(val.to_string())
+        match val {
+            // Preserve HTTP semantics across crates: ApiError can parse [code:404].
+            MegaError::NotFound(msg) => GitError::CustomError(format!("[code:404] {msg}")),
+            MegaError::ObjStorageNotFound(msg) => {
+                GitError::CustomError(format!("[code:404] ObjStorage not found: {msg}"))
+            }
+            other => GitError::CustomError(other.to_string()),
+        }
     }
 }
 

common/src/utils.rs

@@ -7,6 +7,19 @@ pub const ZERO_ID: &str = match std::str::from_utf8(&[b'0'; 40]) {
     Err(_) => panic!("can't get ZERO_ID"),
 };
 
+/// Returns true if `oid` looks like a full Git object id in hex form.
+///
+/// We intentionally only accept full-hex ids here (no short ids), because short ids
+/// require repository-specific disambiguation.
+///
+/// Supported lengths:
+/// - SHA-1   : 40 hex chars
+/// - SHA-256 : 64 hex chars
+pub fn is_full_hex_object_id(oid: &str) -> bool {
+    let is_valid_len = oid.len() == 40 || oid.len() == 64;
+    is_valid_len && oid.as_bytes().iter().all(|b| b.is_ascii_hexdigit())
+}
+
 pub fn generate_id() -> i64 {
     // Call `next_id` to generate a new unique id.
     IdInstance::next_id()
@@ -121,6 +134,42 @@ pub fn get_current_bin_name() -> String {
 mod test {
     use super::*;
 
+    #[test]
+    fn test_is_full_hex_object_id() {
+        // Valid SHA-1 (40 hex)
+        assert!(is_full_hex_object_id(&"0".repeat(40)));
+        assert!(is_full_hex_object_id(&"a".repeat(40)));
+        assert!(is_full_hex_object_id(&"A".repeat(40)));
+        assert!(is_full_hex_object_id(&"f".repeat(40)));
+        assert!(is_full_hex_object_id(&"F".repeat(40)));
+
+        // Valid SHA-256 (64 hex)
+        assert!(is_full_hex_object_id(&"0".repeat(64)));
+        let sha256 = "abcdef".repeat(10) + "abcd"; // 60 + 4 = 64
+        assert!(is_full_hex_object_id(&sha256));
+
+        // Invalid lengths (we don't accept short ids)
+        assert!(!is_full_hex_object_id(""));
+        assert!(!is_full_hex_object_id(&"0".repeat(39)));
+        assert!(!is_full_hex_object_id(&"0".repeat(41)));
+        assert!(!is_full_hex_object_id(&"0".repeat(63)));
+        assert!(!is_full_hex_object_id(&"0".repeat(65)));
+
+        // Invalid characters
+        assert!(!is_full_hex_object_id(
+            &(String::from("g") + &"0".repeat(39))
+        ));
+        assert!(!is_full_hex_object_id(
+            &(String::from("-") + &"0".repeat(39))
+        ));
+        assert!(!is_full_hex_object_id(
+            &(String::from(" ") + &"0".repeat(39))
+        ));
+        assert!(!is_full_hex_object_id(
+            &(String::from("é") + &"0".repeat(39))
+        ));
+    }
+
     #[test]
     fn test_check_conventional_commits() {
         // successfull cases

io-orbit/Cargo.toml

@@ -16,8 +16,9 @@ futures = { workspace = true }
 bytes = { workspace = true }
 async-trait = { workspace = true }
 tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
 thiserror = { workspace = true }
 reqwest = { workspace = true }
-tokio = { workspace = true, features = ["time"] }
+tokio = { workspace = true, features = ["time", "macros", "rt"] }
 serde = { workspace = true }
 serde_json = { workspace = true }

io-orbit/src/adapter.rs

@@ -15,7 +15,7 @@ use reqwest::Method;
 use crate::{
     error::IoOrbitError,
     log_storage::{LogManifest, LogSegmentMeta, LogStorage},
-    object_storage::{MegaObjectStorage, ObjectByteStream, ObjectKey, ObjectMeta},
+    object_storage::{MegaObjectStorage, ObjectByteStream, ObjectKey, ObjectMeta, ObjectNamespace},
 };
 
 /// Strategy used for uploading objects to the underlying [`BackendStore`].
@@ -68,10 +68,14 @@ impl MegaObjectStorage for ObjectStoreAdapter {
         _meta: ObjectMeta,
     ) -> Result<(), MegaError> {
         let path = key.to_object_store_path();
-
-        match self.upload_strategy {
-            UploadStrategy::Multipart => self.put_multipart(&path, data).await,
-            UploadStrategy::SinglePut => self.put_single(&path, data).await,
+        match (key.namespace, &self.upload_strategy) {
+            (ObjectNamespace::Git, UploadStrategy::SinglePut) => {
+                self.put_idempotent(&path, data).await
+            }
+            _ => match self.upload_strategy {
+                UploadStrategy::Multipart => self.put_multipart(&path, data).await,
+                UploadStrategy::SinglePut => self.put_single(&path, data).await,
+            },
         }
     }
 
@@ -692,4 +696,44 @@ impl ObjectStoreAdapter {
 
         Ok(())
     }
+
+    /// Uploads an object using a single `PUT` in **create-only** mode.
+    ///
+    /// This helper is currently used only for **Git objects** (blob/pack data)
+    /// via `put_stream` when `ObjectNamespace::Git` + `UploadStrategy::SinglePut`
+    /// are selected.
+    ///
+    /// Semantics:
+    /// - Uses [`PutMode::Create`], so the backend will fail if the key already exists.
+    /// - This makes writes *idempotent* for content-addressed Git blobs: the first
+    ///   successful upload wins, and later attempts do not silently overwrite data.
+    /// - Callers must ensure that `path` is a content-hash-based key (Git object id),
+    ///   so that "already exists" is expected and safe to ignore at higher layers.
+    async fn put_idempotent(
+        &self,
+        path: &object_store::path::Path,
+        mut data: ObjectByteStream,
+    ) -> Result<(), MegaError> {
+        let mut buf = BytesMut::new();
+        while let Some(chunk) = data.try_next().await? {
+            buf.extend_from_slice(&chunk);
+        }
+
+        // Use `PutMode::Create` so we never overwrite an existing object.
+        // For Git blobs (content-addressed by hash), an "already exists"
+        // error is expected and treated as success by higher layers.
+        match self
+            .to_store()
+            .put_opts(
+                path,
+                PutPayload::from_bytes(buf.into()),
+                PutOptions::from(PutMode::Create),
+            )
+            .await
+        {
+            Ok(_) => Ok(()),
+            Err(object_store::Error::AlreadyExists { .. }) => Ok(()),
+            Err(e) => Err(IoOrbitError::from(e).into()),
+        }
+    }
 }