From 9dd6b2791d04f3680f8dcebd79f671d6f5d687e5 Mon Sep 17 00:00:00 2001 From: pleasedontbreak123 Date: Sat, 14 Feb 2026 21:17:33 +0800 Subject: [PATCH 1/7] feat:clearify object not found erro --- ceres/src/api_service/admin_ops.rs | 18 +++++++-- ceres/src/api_service/commit_ops.rs | 6 ++- ceres/src/api_service/mod.rs | 15 +++++-- ceres/src/lfs/handler.rs | 4 +- common/src/errors.rs | 11 ++++++ common/src/utils.rs | 13 ++++++ io-orbit/src/error.rs | 12 +++++- jupiter/src/service/git_service.rs | 12 ++++++ jupiter/src/storage/mod.rs | 61 +++++++++++++++++++++++++++++ 9 files changed, 140 insertions(+), 12 deletions(-) diff --git a/ceres/src/api_service/admin_ops.rs b/ceres/src/api_service/admin_ops.rs index 9cae01972..5e035dffb 100644 --- a/ceres/src/api_service/admin_ops.rs +++ b/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)))?; diff --git a/ceres/src/api_service/commit_ops.rs b/ceres/src/api_service/commit_ops.rs index 1abadc5ca..62581f0d4 100644 --- a/ceres/src/api_service/commit_ops.rs +++ b/ceres/src/api_service/commit_ops.rs @@ -708,10 +708,12 @@ async fn compute_commit_diff_items( parent_blobs_set.push(blobs); } - let ctx = handler.get_context(); let mut blob_cache: HashMap> = 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); } diff --git a/ceres/src/api_service/mod.rs b/ceres/src/api_service/mod.rs index b8f6893e3..336e58dd6 100644 --- a/ceres/src/api_service/mod.rs +++ b/ceres/src/api_service/mod.rs @@ -122,10 +122,17 @@ pub trait ApiHandler: Send + Sync { ) -> Result; async fn get_raw_blob_by_hash(&self, hash: &str) -> Result, 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 diff --git a/ceres/src/lfs/handler.rs b/ceres/src/lfs/handler.rs index d9931d647..4c08a4ffd 100644 --- a/ceres/src/lfs/handler.rs +++ b/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); diff --git a/common/src/errors.rs b/common/src/errors.rs index b6e246269..bc7904db3 100644 --- a/common/src/errors.rs +++ b/common/src/errors.rs @@ -64,6 +64,17 @@ 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), diff --git a/common/src/utils.rs b/common/src/utils.rs index c7b16bb8f..25427d1af 100644 --- a/common/src/utils.rs +++ b/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() diff --git a/io-orbit/src/error.rs b/io-orbit/src/error.rs index 783c912a6..2424dbaf2 100644 --- a/io-orbit/src/error.rs +++ b/io-orbit/src/error.rs @@ -1,4 +1,5 @@ use common::errors::MegaError; +use object_store::Error as ObjectStoreError; #[derive(Debug, thiserror::Error)] pub enum IoOrbitError { @@ -15,7 +16,16 @@ pub enum IoOrbitError { impl From for MegaError { fn from(err: IoOrbitError) -> Self { match err { - IoOrbitError::ObjectStore(e) => MegaError::ObjStorage(e.to_string()), + IoOrbitError::ObjectStore(e) => match e { + ObjectStoreError::NotFound { .. } => { + // Underlying object store reports that the object/key does not exist. + MegaError::ObjStorageNotFound(e.to_string()) + } + + + // Fallback for any other object_store errors. + _ => MegaError::ObjStorage(e.to_string()), + }, IoOrbitError::WriteManifestPreconditionFailed => { MegaError::Other("write manifest precondition failed".to_string()) } diff --git a/jupiter/src/service/git_service.rs b/jupiter/src/service/git_service.rs index 23db8a919..ea3205254 100644 --- a/jupiter/src/service/git_service.rs +++ b/jupiter/src/service/git_service.rs @@ -1,5 +1,6 @@ use bytes::Bytes; use common::errors::MegaError; +use common::utils::is_full_hex_object_id; use futures::StreamExt; use git_internal::internal::object::blob::Blob; use io_orbit::{ @@ -83,6 +84,11 @@ impl GitService { } pub async fn get_object_as_bytes(&self, hash: &str) -> Result, MegaError> { + // Avoid sending obviously invalid object ids to object storage backends. + if !is_full_hex_object_id(hash) { + return Err(MegaError::Other("Invalid object ID format".to_string())); + } + let key = ObjectKey { namespace: ObjectNamespace::Git, key: hash.to_string(), @@ -100,6 +106,12 @@ impl GitService { } pub fn get_objects_stream(&self, hashes: Vec) -> MultiObjectByteStream<'_> { + // Filter out obviously invalid object ids early to avoid spurious backend requests. + // Callers that need strict validation should validate up-front and return 4xx. + let hashes = hashes + .into_iter() + .filter(|h| is_full_hex_object_id(h)) + .collect::>(); self.obj_storage.inner.get_many( hashes .into_iter() diff --git a/jupiter/src/storage/mod.rs b/jupiter/src/storage/mod.rs index 7b8688456..f7da904a1 100644 --- a/jupiter/src/storage/mod.rs +++ b/jupiter/src/storage/mod.rs @@ -270,6 +270,67 @@ impl Storage { self.app_service.mono_storage.clone() } + /// Best-effort classification/logging helper for object storage "not found" errors + /// when dealing with Git blobs. + /// + /// Behavior: + /// - If `err` is not `MegaError::ObjStorageNotFound`, it is returned unchanged. + /// - If it is `ObjStorageNotFound(msg)`, we: + /// - Look up blob metadata via `mega_blob` table + /// - Emit structured logs to distinguish: + /// - "missing_in_db_and_s3" (likely never written / invalid request) + /// - "missing_in_s3_but_has_meta" (data loss or external deletion) + /// - Return a new `MegaError::ObjStorageNotFound` whose message is prefixed with + /// a classification tag: + /// - "[obj_missing_in_db_and_s3] ..." + /// - "[obj_missing_in_s3_but_has_meta] ..." + /// - "[obj_meta_lookup_failed] ..." (if metadata lookup itself fails) + pub async fn classify_blob_objstorage_not_found( + &self, + hash: &str, + err: MegaError, + ) -> MegaError { + match err { + MegaError::ObjStorageNotFound(msg) => { + let mono_storage = self.mono_storage(); + + + match mono_storage + .get_mega_blobs_by_hashes(vec![hash.to_string()]) + .await + { + Ok(blobs) if blobs.is_empty() => { + + tracing::warn!( + "Blob {hash} not found in both object storage and metadata (likely never written or invalid request)" + ); + return MegaError::ObjStorageNotFound(msg); + } + Ok(mut blobs) => { + + if let Some(blob) = blobs.pop() { + tracing::error!( + "Object { } missing in S3 but metadata exists; possible data loss or misconfiguration", + blob.blob_id + ); + return MegaError::ObjStorageInconsistent(format!("Object{hash} missing in S3 but metadata exists; possible data loss or misconfiguration ")); + } else { + tracing::error!( + "Object missing in S3 but metadata lookup returned unexpected empty result", + ); + return MegaError::ObjStorageInconsistent(format!("Object {hash} missing in S3 but metadata lookup returned unexpected empty result ")); + } + } + Err(_) => { + return MegaError::ObjStorageInconsistent(format!("Failed to query blob {hash} metadata while handling ObjStorageNotFound ")); + } + } + + } + other => other, + } + } + pub fn git_db_storage(&self) -> GitDbStorage { self.app_service.git_db_storage.clone() } From 4931ecb204d87581e1ab2d539077c01347128d5c Mon Sep 17 00:00:00 2001 From: nbei Date: Mon, 23 Feb 2026 20:41:04 +0800 Subject: [PATCH 2/7] fix: handle data consistency issues in a simple way --- io-orbit/src/adapter.rs | 36 +++++++++++++++++++++++++---- io-orbit/src/object_storage.rs | 13 ++++++++++- jupiter/src/service/mono_service.rs | 6 +++-- 3 files changed, 47 insertions(+), 8 deletions(-) diff --git a/io-orbit/src/adapter.rs b/io-orbit/src/adapter.rs index 876a5e7de..f568ad289 100644 --- a/io-orbit/src/adapter.rs +++ b/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,11 +68,18 @@ 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, + } + } } + } async fn get_stream( @@ -692,4 +699,23 @@ impl ObjectStoreAdapter { Ok(()) } + + 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); + } + + self + .to_store() + .put_opts(&path, PutPayload::from_bytes(buf.into()), PutOptions::from(PutMode::Create)) + .await + .map_err(IoOrbitError::from)?; + Ok(()) + } } diff --git a/io-orbit/src/object_storage.rs b/io-orbit/src/object_storage.rs index a0d8cf1b1..9dfbe26dd 100644 --- a/io-orbit/src/object_storage.rs +++ b/io-orbit/src/object_storage.rs @@ -214,7 +214,18 @@ pub trait MegaObjectStorage: Send + Sync { ) -> Result<(), MegaError> { objects .try_for_each_concurrent(concurrency, |(key, stream, meta)| async move { - self.put_stream(&key, stream, meta).await + self.put_stream(&key, stream, meta) + .await + // Attach key/context so callers can see *which* object failed + // instead of only getting a bare storage error. This avoids + .map_err(|e| { + MegaError::Other(format!( + "object_storage::put_many failed for key={} namespace={} error_chain:\n{}", + key.key, + key.namespace, + dump_error_chain(&e), + )) + }) }) .await } diff --git a/jupiter/src/service/mono_service.rs b/jupiter/src/service/mono_service.rs index 537287354..deedfb546 100644 --- a/jupiter/src/service/mono_service.rs +++ b/jupiter/src/service/mono_service.rs @@ -154,6 +154,8 @@ impl MonoService { Ok(git_objects.commits) } + /// Writes blobs to object storage (S3) first, then to DB. + /// This order avoids leaving blob rows in DB when S3 write fails (same as save_entry). pub async fn save_blobs(&self, commit_id: &str, blobs: Vec) -> Result<(), MegaError> { let mega_blobs: Vec = blobs .iter() @@ -163,7 +165,7 @@ impl MonoService { m.into_active_model() }) .collect(); - self.mono_storage.batch_save_model(mega_blobs).await?; - self.git_service.put_objects(blobs).await + self.git_service.put_objects(blobs).await?; + self.mono_storage.batch_save_model(mega_blobs).await } } From d3268f1701a8dde174067ec402c2e0050b0531c5 Mon Sep 17 00:00:00 2001 From: nbei Date: Tue, 24 Feb 2026 15:28:27 +0800 Subject: [PATCH 3/7] fix api panic --- ceres/src/api_service/mono_api_service.rs | 28 +- common/src/errors.rs | 9 +- io-orbit/Cargo.toml | 3 +- io-orbit/src/bin/migrate_local_to_s3.rs | 302 ++++++++++++++++++++++ jupiter/src/storage/mod.rs | 12 +- mono/src/api/api_router.rs | 13 +- mono/src/api/error.rs | 1 + 7 files changed, 338 insertions(+), 30 deletions(-) create mode 100644 io-orbit/src/bin/migrate_local_to_s3.rs diff --git a/ceres/src/api_service/mono_api_service.rs b/ceres/src/api_service/mono_api_service.rs index 8fc1fece0..dc26af1a0 100644 --- a/ceres/src/api_service/mono_api_service.rs +++ b/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 { - 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 { - 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( diff --git a/common/src/errors.rs b/common/src/errors.rs index bc7904db3..fb2ba699c 100644 --- a/common/src/errors.rs +++ b/common/src/errors.rs @@ -94,7 +94,14 @@ impl From for MegaError { impl From 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()), + } } } diff --git a/io-orbit/Cargo.toml b/io-orbit/Cargo.toml index bbdb83fe5..40def9e07 100644 --- a/io-orbit/Cargo.toml +++ b/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 } diff --git a/io-orbit/src/bin/migrate_local_to_s3.rs b/io-orbit/src/bin/migrate_local_to_s3.rs new file mode 100644 index 000000000..af82b5d08 --- /dev/null +++ b/io-orbit/src/bin/migrate_local_to_s3.rs @@ -0,0 +1,302 @@ +//! Offline object migration: LocalFileSystem → Amazon S3. +//! +//! This binary scans a **local object storage directory** (the same layout used by Mega's +//! `ObjectNamespace::{Git,Lfs,Log}` sharding paths) and uploads all objects to an **S3 bucket**. +//! +//! It is intended for **one-time / offline** backfill or migration jobs (e.g. moving from +//! the local backend to S3). It is *not* a continuous sync tool. +//! +//! ## What it does +//! - Lists all objects under the configured local root directory. +//! - For each object: +//! - Performs an S3 `HEAD` request and **skips** the upload if the object already exists. +//! - Otherwise reads the local object and uploads it to S3. +//! - Runs uploads concurrently (configurable). +//! +//! ## What it does NOT do +//! - It does **not delete** local objects after upload. +//! - It does **not overwrite** objects on S3 (it skips when `HEAD` succeeds). +//! - It does **not** perform checksums/ETag validation between local and S3. +//! - It does **not** retry uploads; any failure aborts the migration. +//! +//! ## Safety / idempotency +//! The migration is **idempotent** in the sense that re-running it will skip any objects that +//! already exist on S3 (based on `HEAD` success). This avoids accidental overwrites. +//! If you need strict "exists but different content" detection, add a `head`-then-compare +//! strategy (size/etag/checksum) before skipping. +//! +//! ## Performance notes +//! - Concurrency is controlled via the `MIGRATE_CONCURRENCY` environment variable. +//! - Objects are uploaded using **multipart streaming** to avoid buffering the full object in memory. +//! - The Tokio runtime is `current_thread`; concurrency here is mostly I/O concurrency. +//! +//! ## Configuration +//! This tool reads Mega config and uses: +//! - `object_storage.local.root_dir` as the source directory +//! - `object_storage.s3.{region,bucket,access_key_id,secret_access_key}` as the destination +//! +//! The config path resolution order is: +//! 1. CLI: `--config ` +//! 2. Environment: `MEGA_CONFIG=` +//! +//! ## Usage +//! +//! From the workspace root: +//! +//! ```bash +//! # Basic (explicit config path) +//! cargo run -p io-orbit --bin migrate_local_to_s3 -- --config ./config.toml +//! +//! # Or via env var +//! MEGA_CONFIG=./config.toml cargo run -p io-orbit --bin migrate_local_to_s3 +//! +//! # Tune concurrency (default: 16) +//! MIGRATE_CONCURRENCY=32 MEGA_CONFIG=./config.toml cargo run -p io-orbit --bin migrate_local_to_s3 +//! ``` +//! +//! ## Verification tips +//! - Watch logs for `skip existing object on s3:` and `migrating object:` lines. +//! - Consider checking a sample of keys on S3 (e.g. `aws s3api head-object`) after completion. +//! - If you run S3-compatible storage (MinIO/rustfs), this binary currently uses the AWS S3 +//! builder without a custom endpoint; adapt `AmazonS3Builder` accordingly if needed. + +use std::path::PathBuf; +use std::sync::Arc; + +use common::config::{loader::{ConfigInput, ConfigLoader}, Config, ObjectStorageBackend}; +use common::errors::MegaError; +use futures::{StreamExt, TryStreamExt}; +use object_store::{ + aws::AmazonS3Builder, local::LocalFileSystem, ObjectStore, ObjectStoreExt, +}; +use tokio::sync::Semaphore; +use tracing::{error, info}; + +#[tokio::main(flavor = "current_thread")] +async fn main() { + if let Err(e) = run().await { + eprintln!("migration failed: {}", e); + std::process::exit(1); + } +} + +async fn run() -> Result<(), MegaError> { + init_tracing(); + + let config_path = load_config_path()?; + let config = Config::new( + config_path + .to_str() + .ok_or_else(|| MegaError::Other("config path is not valid UTF-8".to_string()))?, + )?; + + let object_cfg = config.object_storage.clone(); + let backend = config.monorepo.storage_type; + + // Build Local backend (source) + let local_root = object_cfg.local.root_dir.clone(); + let local = LocalFileSystem::new_with_prefix(&local_root) + .map_err(|e| MegaError::Other(format!("failed to init LocalFileSystem: {e}")))?; + let local = Arc::new(local); + + // Build S3 backend (target) according to monorepo.storage_type: + // - S3 -> real AWS S3 (no custom endpoint) + // - S3Compatible -> S3-compatible service (RustFS/MinIO etc.) using endpoint_url + let s3_cfg = object_cfg.s3.clone(); + let s3 = match backend { + ObjectStorageBackend::S3 => { + AmazonS3Builder::new() + .with_region(&s3_cfg.region) + .with_bucket_name(&s3_cfg.bucket) + .with_access_key_id(&s3_cfg.access_key_id) + .with_secret_access_key(&s3_cfg.secret_access_key) + .build() + .map_err(|e| MegaError::Other(format!("failed to init S3 client: {e}")))? + } + ObjectStorageBackend::S3Compatible => { + AmazonS3Builder::new() + .with_region(&s3_cfg.region) + .with_bucket_name(&s3_cfg.bucket) + .with_access_key_id(&s3_cfg.access_key_id) + .with_secret_access_key(&s3_cfg.secret_access_key) + .with_endpoint(&s3_cfg.endpoint_url) + .with_allow_http(true) + .with_virtual_hosted_style_request(false) + .build() + .map_err(|e| { + MegaError::Other(format!("failed to init S3-compatible client: {e}")) + })? + } + other => { + return Err(MegaError::Other(format!( + "migrate_local_to_s3 only supports S3/S3Compatible targets, got {:?}", + other + ))); + } + }; + let s3 = Arc::new(s3); + + info!( + "starting offline migration from local {:?} to s3 bucket {:?}", + local_root, s3_cfg.bucket + ); + + // Concurrency for in-flight uploads. Higher isn't always better: S3 request rate limits + // and local disk throughput will cap effective concurrency. + let concurrency = std::env::var("MIGRATE_CONCURRENCY") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|&v| v > 0) + .unwrap_or(16); + + migrate_all(local, s3, concurrency).await +} + +fn init_tracing() { + // Simple stderr logger; we don't depend on global app logging here. + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::Level::INFO) + .with_target(true) + .try_init(); +} + +fn load_config_path() -> Result { + let mut cli_path: Option = None; + + // Very lightweight CLI parsing: + // - supports optional `--config ` + // - ignores unknown args (prints a warning) + let mut args = std::env::args().skip(1); + while let Some(arg) = args.next() { + if arg == "--config" { + if let Some(p) = args.next() { + cli_path = Some(PathBuf::from(p)); + } + } else { + eprintln!("warning: unknown argument `{}` is ignored", arg); + } + } + + let input = ConfigInput { + cli_path, + env_path: std::env::var_os("MEGA_CONFIG").map(PathBuf::from), + }; + + let loaded = ConfigLoader::new(input) + .load() + .map_err(|e| MegaError::Other(format!("failed to load config path: {e}")))?; + + Ok(loaded.path) +} + +async fn migrate_all( + local: Arc, + s3: Arc, + concurrency: usize, +) -> Result<(), MegaError> { + let semaphore = Arc::new(Semaphore::new(concurrency)); + + // List everything under the local root. Passing `None` means "from the root prefix". + let mut listed = local.list(None); + let mut tasks = Vec::new(); + + while let Some(entry) = listed.next().await { + let meta = match entry { + Ok(m) => m, + Err(e) => { + error!("list local objects failed: {e}"); + return Err(MegaError::Other(format!( + "list local objects failed: {e}" + ))); + } + }; + + let path = meta.location.clone(); + let permit = semaphore + .clone() + .acquire_owned() + .await + .map_err(|e| MegaError::Other(format!("failed to acquire semaphore: {e}")))?; + + let local_clone = local.clone(); + let s3_clone = s3.clone(); + + tasks.push(tokio::spawn(async move { + let _permit = permit; + + // Skip if already exists on S3. + // + // NOTE: + // - This makes the migration re-runnable and avoids overwriting. + // - If the object exists but has different content, this tool will not detect it. + // Add a `head`-then-compare strategy if you need that guarantee. + if s3_clone.head(&path).await.is_ok() { + info!("skip existing object on s3: {}", path); + return Ok::<(), MegaError>(()); + } + + info!("migrating object: {}", path); + + // Read the local object. + let obj = local_clone.get(&path).await.map_err(|e| { + MegaError::Other(format!("failed to read local object {}: {e}", path)) + })?; + + // Stream upload to S3 using multipart. + // + // Why multipart: + // - avoids buffering the whole object in memory + // - works well for large objects + // + // Notes: + // - if any part upload fails, we abort the multipart upload to avoid leaving + // dangling uploads on the backend. + let mut upload = s3_clone.put_multipart(&path).await.map_err(|e| { + MegaError::Other(format!("failed to init multipart upload {}: {e}", path)) + })?; + + let mut stream = obj.into_stream(); + let res: Result<(), MegaError> = async { + while let Some(chunk) = stream.try_next().await.map_err(|e| { + MegaError::Other(format!("failed to read local stream chunk {}: {e}", path)) + })? { + upload.put_part(chunk.into()).await.map_err(|e| { + MegaError::Other(format!( + "failed to upload multipart part for {}: {e}", + path + )) + })?; + } + + upload.complete().await.map_err(|e| { + MegaError::Other(format!("failed to complete multipart upload {}: {e}", path)) + })?; + + Ok(()) + } + .await; + + if let Err(ref e) = res { + // Best-effort abort; if abort fails, surface abort error as well. + if let Err(abort_err) = upload.abort().await { + return Err(MegaError::Other(format!( + "multipart upload failed for {}: {}; abort also failed: {}", + path, e, abort_err + ))); + } + } + res?; + + Ok(()) + })); + } + + for t in tasks { + t.await + .map_err(|e| MegaError::Other(format!("migration task panicked: {e}")))??; + } + + info!("offline migration completed"); + Ok(()) +} + diff --git a/jupiter/src/storage/mod.rs b/jupiter/src/storage/mod.rs index f7da904a1..4fa2b26ee 100644 --- a/jupiter/src/storage/mod.rs +++ b/jupiter/src/storage/mod.rs @@ -291,7 +291,7 @@ impl Storage { err: MegaError, ) -> MegaError { match err { - MegaError::ObjStorageNotFound(msg) => { + MegaError::ObjStorageNotFound(_msg) => { let mono_storage = self.mono_storage(); @@ -300,22 +300,22 @@ impl Storage { .await { Ok(blobs) if blobs.is_empty() => { - - tracing::warn!( + let friendly = format!( "Blob {hash} not found in both object storage and metadata (likely never written or invalid request)" ); - return MegaError::ObjStorageNotFound(msg); + tracing::warn!("{}", friendly); + return MegaError::ObjStorageNotFound(friendly); } Ok(mut blobs) => { if let Some(blob) = blobs.pop() { - tracing::error!( + tracing::warn!( "Object { } missing in S3 but metadata exists; possible data loss or misconfiguration", blob.blob_id ); return MegaError::ObjStorageInconsistent(format!("Object{hash} missing in S3 but metadata exists; possible data loss or misconfiguration ")); } else { - tracing::error!( + tracing::warn!( "Object missing in S3 but metadata lookup returned unexpected empty result", ); return MegaError::ObjStorageInconsistent(format!("Object {hash} missing in S3 but metadata lookup returned unexpected empty result ")); diff --git a/mono/src/api/api_router.rs b/mono/src/api/api_router.rs index b2a2ecf7c..753a68eed 100644 --- a/mono/src/api/api_router.rs +++ b/mono/src/api/api_router.rs @@ -1,4 +1,4 @@ -use anyhow::Result; +use anyhow::{anyhow, Result}; use axum::{ Json, body::Body, @@ -7,7 +7,6 @@ use axum::{ routing::get, }; use ceres::{api_service::ApiHandler, model::git::TreeQuery}; -use http::StatusCode; use utoipa_axum::{router::OpenApiRouter, routes}; use crate::{ @@ -78,12 +77,10 @@ pub async fn get_blob_file( .header("Content-Disposition", file_name) .body(Body::from(data)) .unwrap()), - Err(_) => Ok({ - Response::builder() - .status(StatusCode::NOT_FOUND) - .body(Body::empty()) - .unwrap() - }), + Err(e) => Err(ApiError::not_found(anyhow!( + "error={}", + e + ))), } } diff --git a/mono/src/api/error.rs b/mono/src/api/error.rs index 80a7c3d7d..994cac684 100644 --- a/mono/src/api/error.rs +++ b/mono/src/api/error.rs @@ -136,6 +136,7 @@ where // Handle other MegaError variants match mega_err { + MegaError::NotFound(_) => return ApiError::not_found(anyhow_err), MegaError::Db(_) | MegaError::Redis(_) | MegaError::Io(_) => { // Hide internal details in production, return generic 500 tracing::error!( From 6defe8e173ca79c716fd37712d610ea907441a90 Mon Sep 17 00:00:00 2001 From: nbei Date: Tue, 24 Feb 2026 15:43:06 +0800 Subject: [PATCH 4/7] fix clippy --- ceres/src/api_service/mod.rs | 4 +- common/src/errors.rs | 2 - io-orbit/src/adapter.rs | 25 +++++----- io-orbit/src/bin/migrate_local_to_s3.rs | 63 +++++++++++-------------- io-orbit/src/error.rs | 3 +- jupiter/src/service/git_service.rs | 3 +- jupiter/src/storage/mod.rs | 17 ++++--- mono/src/api/api_router.rs | 7 +-- 8 files changed, 55 insertions(+), 69 deletions(-) diff --git a/ceres/src/api_service/mod.rs b/ceres/src/api_service/mod.rs index 336e58dd6..a44f04223 100644 --- a/ceres/src/api_service/mod.rs +++ b/ceres/src/api_service/mod.rs @@ -127,9 +127,7 @@ pub trait ApiHandler: Send + Sync { Ok(data) => Ok(data), Err(e) => { // Best-effort classification/logging for ObjStorageNotFound cases. - let e = storage - .classify_blob_objstorage_not_found(hash, e) - .await; + let e = storage.classify_blob_objstorage_not_found(hash, e).await; Err(e) } } diff --git a/common/src/errors.rs b/common/src/errors.rs index fb2ba699c..17cbd9b51 100644 --- a/common/src/errors.rs +++ b/common/src/errors.rs @@ -72,8 +72,6 @@ pub enum MegaError { ///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}")] diff --git a/io-orbit/src/adapter.rs b/io-orbit/src/adapter.rs index f568ad289..13af7925f 100644 --- a/io-orbit/src/adapter.rs +++ b/io-orbit/src/adapter.rs @@ -72,14 +72,11 @@ impl MegaObjectStorage for ObjectStoreAdapter { (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, - } - } + _ => match self.upload_strategy { + UploadStrategy::Multipart => self.put_multipart(&path, data).await, + UploadStrategy::SinglePut => self.put_single(&path, data).await, + }, } - } async fn get_stream( @@ -700,20 +697,22 @@ impl ObjectStoreAdapter { Ok(()) } - async fn put_idempotent( + 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); } - - self - .to_store() - .put_opts(&path, PutPayload::from_bytes(buf.into()), PutOptions::from(PutMode::Create)) + + self.to_store() + .put_opts( + path, + PutPayload::from_bytes(buf.into()), + PutOptions::from(PutMode::Create), + ) .await .map_err(IoOrbitError::from)?; Ok(()) diff --git a/io-orbit/src/bin/migrate_local_to_s3.rs b/io-orbit/src/bin/migrate_local_to_s3.rs index af82b5d08..0e89759f8 100644 --- a/io-orbit/src/bin/migrate_local_to_s3.rs +++ b/io-orbit/src/bin/migrate_local_to_s3.rs @@ -60,15 +60,17 @@ //! - If you run S3-compatible storage (MinIO/rustfs), this binary currently uses the AWS S3 //! builder without a custom endpoint; adapt `AmazonS3Builder` accordingly if needed. -use std::path::PathBuf; -use std::sync::Arc; - -use common::config::{loader::{ConfigInput, ConfigLoader}, Config, ObjectStorageBackend}; -use common::errors::MegaError; -use futures::{StreamExt, TryStreamExt}; -use object_store::{ - aws::AmazonS3Builder, local::LocalFileSystem, ObjectStore, ObjectStoreExt, +use std::{path::PathBuf, sync::Arc}; + +use common::{ + config::{ + Config, ObjectStorageBackend, + loader::{ConfigInput, ConfigLoader}, + }, + errors::MegaError, }; +use futures::{StreamExt, TryStreamExt}; +use object_store::{ObjectStore, ObjectStoreExt, aws::AmazonS3Builder, local::LocalFileSystem}; use tokio::sync::Semaphore; use tracing::{error, info}; @@ -104,29 +106,23 @@ async fn run() -> Result<(), MegaError> { // - S3Compatible -> S3-compatible service (RustFS/MinIO etc.) using endpoint_url let s3_cfg = object_cfg.s3.clone(); let s3 = match backend { - ObjectStorageBackend::S3 => { - AmazonS3Builder::new() - .with_region(&s3_cfg.region) - .with_bucket_name(&s3_cfg.bucket) - .with_access_key_id(&s3_cfg.access_key_id) - .with_secret_access_key(&s3_cfg.secret_access_key) - .build() - .map_err(|e| MegaError::Other(format!("failed to init S3 client: {e}")))? - } - ObjectStorageBackend::S3Compatible => { - AmazonS3Builder::new() - .with_region(&s3_cfg.region) - .with_bucket_name(&s3_cfg.bucket) - .with_access_key_id(&s3_cfg.access_key_id) - .with_secret_access_key(&s3_cfg.secret_access_key) - .with_endpoint(&s3_cfg.endpoint_url) - .with_allow_http(true) - .with_virtual_hosted_style_request(false) - .build() - .map_err(|e| { - MegaError::Other(format!("failed to init S3-compatible client: {e}")) - })? - } + ObjectStorageBackend::S3 => AmazonS3Builder::new() + .with_region(&s3_cfg.region) + .with_bucket_name(&s3_cfg.bucket) + .with_access_key_id(&s3_cfg.access_key_id) + .with_secret_access_key(&s3_cfg.secret_access_key) + .build() + .map_err(|e| MegaError::Other(format!("failed to init S3 client: {e}")))?, + ObjectStorageBackend::S3Compatible => AmazonS3Builder::new() + .with_region(&s3_cfg.region) + .with_bucket_name(&s3_cfg.bucket) + .with_access_key_id(&s3_cfg.access_key_id) + .with_secret_access_key(&s3_cfg.secret_access_key) + .with_endpoint(&s3_cfg.endpoint_url) + .with_allow_http(true) + .with_virtual_hosted_style_request(false) + .build() + .map_err(|e| MegaError::Other(format!("failed to init S3-compatible client: {e}")))?, other => { return Err(MegaError::Other(format!( "migrate_local_to_s3 only supports S3/S3Compatible targets, got {:?}", @@ -205,9 +201,7 @@ async fn migrate_all( Ok(m) => m, Err(e) => { error!("list local objects failed: {e}"); - return Err(MegaError::Other(format!( - "list local objects failed: {e}" - ))); + return Err(MegaError::Other(format!("list local objects failed: {e}"))); } }; @@ -299,4 +293,3 @@ async fn migrate_all( info!("offline migration completed"); Ok(()) } - diff --git a/io-orbit/src/error.rs b/io-orbit/src/error.rs index 2424dbaf2..a468a7301 100644 --- a/io-orbit/src/error.rs +++ b/io-orbit/src/error.rs @@ -21,8 +21,7 @@ impl From for MegaError { // Underlying object store reports that the object/key does not exist. MegaError::ObjStorageNotFound(e.to_string()) } - - + // Fallback for any other object_store errors. _ => MegaError::ObjStorage(e.to_string()), }, diff --git a/jupiter/src/service/git_service.rs b/jupiter/src/service/git_service.rs index ea3205254..4a089a1a7 100644 --- a/jupiter/src/service/git_service.rs +++ b/jupiter/src/service/git_service.rs @@ -1,6 +1,5 @@ use bytes::Bytes; -use common::errors::MegaError; -use common::utils::is_full_hex_object_id; +use common::{errors::MegaError, utils::is_full_hex_object_id}; use futures::StreamExt; use git_internal::internal::object::blob::Blob; use io_orbit::{ diff --git a/jupiter/src/storage/mod.rs b/jupiter/src/storage/mod.rs index 4fa2b26ee..6e0b9a7cb 100644 --- a/jupiter/src/storage/mod.rs +++ b/jupiter/src/storage/mod.rs @@ -293,7 +293,6 @@ impl Storage { match err { MegaError::ObjStorageNotFound(_msg) => { let mono_storage = self.mono_storage(); - match mono_storage .get_mega_blobs_by_hashes(vec![hash.to_string()]) @@ -304,28 +303,32 @@ impl Storage { "Blob {hash} not found in both object storage and metadata (likely never written or invalid request)" ); tracing::warn!("{}", friendly); - return MegaError::ObjStorageNotFound(friendly); + MegaError::ObjStorageNotFound(friendly) } Ok(mut blobs) => { - if let Some(blob) = blobs.pop() { tracing::warn!( "Object { } missing in S3 but metadata exists; possible data loss or misconfiguration", blob.blob_id ); - return MegaError::ObjStorageInconsistent(format!("Object{hash} missing in S3 but metadata exists; possible data loss or misconfiguration ")); + MegaError::ObjStorageInconsistent(format!( + "Object{hash} missing in S3 but metadata exists; possible data loss or misconfiguration " + )) } else { tracing::warn!( "Object missing in S3 but metadata lookup returned unexpected empty result", ); - return MegaError::ObjStorageInconsistent(format!("Object {hash} missing in S3 but metadata lookup returned unexpected empty result ")); + MegaError::ObjStorageInconsistent(format!( + "Object {hash} missing in S3 but metadata lookup returned unexpected empty result " + )) } } Err(_) => { - return MegaError::ObjStorageInconsistent(format!("Failed to query blob {hash} metadata while handling ObjStorageNotFound ")); + MegaError::ObjStorageInconsistent(format!( + "Failed to query blob {hash} metadata while handling ObjStorageNotFound " + )) } } - } other => other, } diff --git a/mono/src/api/api_router.rs b/mono/src/api/api_router.rs index 753a68eed..f158ba973 100644 --- a/mono/src/api/api_router.rs +++ b/mono/src/api/api_router.rs @@ -1,4 +1,4 @@ -use anyhow::{anyhow, Result}; +use anyhow::{Result, anyhow}; use axum::{ Json, body::Body, @@ -77,10 +77,7 @@ pub async fn get_blob_file( .header("Content-Disposition", file_name) .body(Body::from(data)) .unwrap()), - Err(e) => Err(ApiError::not_found(anyhow!( - "error={}", - e - ))), + Err(e) => Err(ApiError::not_found(anyhow!("error={}", e))), } } From 263f6475ef06a55f0cc825aaff51f283db2cddbd Mon Sep 17 00:00:00 2001 From: nbei Date: Tue, 24 Feb 2026 16:11:36 +0800 Subject: [PATCH 5/7] fix fmt --- jupiter/src/storage/mod.rs | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/jupiter/src/storage/mod.rs b/jupiter/src/storage/mod.rs index 6e0b9a7cb..1e09b13fc 100644 --- a/jupiter/src/storage/mod.rs +++ b/jupiter/src/storage/mod.rs @@ -323,11 +323,9 @@ impl Storage { )) } } - Err(_) => { - MegaError::ObjStorageInconsistent(format!( - "Failed to query blob {hash} metadata while handling ObjStorageNotFound " - )) - } + Err(_) => MegaError::ObjStorageInconsistent(format!( + "Failed to query blob {hash} metadata while handling ObjStorageNotFound " + )), } } other => other, From 4a511db73cd1af65d4acc1bace6963b5f21834d0 Mon Sep 17 00:00:00 2001 From: nbei Date: Tue, 24 Feb 2026 16:44:32 +0800 Subject: [PATCH 6/7] fix Leftover --- common/src/utils.rs | 36 +++++++++++++++++++++++++ io-orbit/src/adapter.rs | 12 +++++++++ io-orbit/src/bin/migrate_local_to_s3.rs | 13 +++++++++ jupiter/src/service/mono_service.rs | 6 ++--- jupiter/src/storage/mod.rs | 30 +++++++++------------ mono/src/api/api_router.rs | 6 ++++- 6 files changed, 80 insertions(+), 23 deletions(-) diff --git a/common/src/utils.rs b/common/src/utils.rs index 25427d1af..028f9d3e5 100644 --- a/common/src/utils.rs +++ b/common/src/utils.rs @@ -134,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 diff --git a/io-orbit/src/adapter.rs b/io-orbit/src/adapter.rs index 13af7925f..d113b1809 100644 --- a/io-orbit/src/adapter.rs +++ b/io-orbit/src/adapter.rs @@ -697,6 +697,18 @@ 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, diff --git a/io-orbit/src/bin/migrate_local_to_s3.rs b/io-orbit/src/bin/migrate_local_to_s3.rs index 0e89759f8..e134ae0e4 100644 --- a/io-orbit/src/bin/migrate_local_to_s3.rs +++ b/io-orbit/src/bin/migrate_local_to_s3.rs @@ -194,6 +194,9 @@ async fn migrate_all( // List everything under the local root. Passing `None` means "from the root prefix". let mut listed = local.list(None); + // Bounded collection of join handles to avoid unbounded memory growth when + // migrating very large object stores. We cap the number of stored handles + // to a small multiple of the concurrency limit. let mut tasks = Vec::new(); while let Some(entry) = listed.next().await { @@ -283,6 +286,16 @@ async fn migrate_all( Ok(()) })); + + // Periodically await some tasks to keep the number of in-memory + // JoinHandles bounded. Semaphore still enforces the true I/O + // concurrency; this only caps bookkeeping overhead. + if tasks.len() >= concurrency.saturating_mul(4).max(64) { + if let Some(t) = tasks.pop() { + t.await + .map_err(|e| MegaError::Other(format!("migration task panicked: {e}")))??; + } + } } for t in tasks { diff --git a/jupiter/src/service/mono_service.rs b/jupiter/src/service/mono_service.rs index deedfb546..b6f7cbf40 100644 --- a/jupiter/src/service/mono_service.rs +++ b/jupiter/src/service/mono_service.rs @@ -70,13 +70,11 @@ impl MonoService { .batch_save_model_with_txn(mega_trees, Some(&txn)) .await?; let mega_blobs = converter.mega_blobs.borrow().values().cloned().collect(); + let raw_blobs = converter.raw_blobs.into_inner(); + self.git_service.put_objects(raw_blobs).await?; self.mono_storage .batch_save_model_with_txn(mega_blobs, Some(&txn)) .await?; - - self.git_service - .put_objects(converter.raw_blobs.into_inner()) - .await?; Ok(txn.commit().await?) } diff --git a/jupiter/src/storage/mod.rs b/jupiter/src/storage/mod.rs index 1e09b13fc..0ce657daa 100644 --- a/jupiter/src/storage/mod.rs +++ b/jupiter/src/storage/mod.rs @@ -300,31 +300,25 @@ impl Storage { { Ok(blobs) if blobs.is_empty() => { let friendly = format!( - "Blob {hash} not found in both object storage and metadata (likely never written or invalid request)" + "[obj_missing_in_db_and_s3] Blob {hash} not found in both object storage and metadata (likely never written or invalid request)" ); tracing::warn!("{}", friendly); MegaError::ObjStorageNotFound(friendly) } Ok(mut blobs) => { - if let Some(blob) = blobs.pop() { - tracing::warn!( - "Object { } missing in S3 but metadata exists; possible data loss or misconfiguration", - blob.blob_id - ); - MegaError::ObjStorageInconsistent(format!( - "Object{hash} missing in S3 but metadata exists; possible data loss or misconfiguration " - )) - } else { - tracing::warn!( - "Object missing in S3 but metadata lookup returned unexpected empty result", - ); - MegaError::ObjStorageInconsistent(format!( - "Object {hash} missing in S3 but metadata lookup returned unexpected empty result " - )) - } + let blob = blobs.pop().expect( + "blobs is guaranteed non-empty here due to match guard on previous arm", + ); + tracing::error!( + "[obj_missing_in_s3_but_has_meta] Object with hash {hash} missing in S3 but metadata exists in DB for blob_id {}; possible data loss or misconfiguration", + blob.blob_id, + ); + MegaError::ObjStorageInconsistent(format!( + "[obj_missing_in_s3_but_has_meta] Object {hash} missing in S3 but metadata exists; possible data loss or misconfiguration" + )) } Err(_) => MegaError::ObjStorageInconsistent(format!( - "Failed to query blob {hash} metadata while handling ObjStorageNotFound " + "[obj_meta_lookup_failed] Failed to query blob {hash} metadata while handling ObjStorageNotFound" )), } } diff --git a/mono/src/api/api_router.rs b/mono/src/api/api_router.rs index f158ba973..81f7a3933 100644 --- a/mono/src/api/api_router.rs +++ b/mono/src/api/api_router.rs @@ -7,6 +7,7 @@ use axum::{ routing::get, }; use ceres::{api_service::ApiHandler, model::git::TreeQuery}; +use common::errors::MegaError; use utoipa_axum::{router::OpenApiRouter, routes}; use crate::{ @@ -77,7 +78,10 @@ pub async fn get_blob_file( .header("Content-Disposition", file_name) .body(Body::from(data)) .unwrap()), - Err(e) => Err(ApiError::not_found(anyhow!("error={}", e))), + Err(e) => match e { + MegaError::ObjStorageNotFound(_) => Err(ApiError::not_found(anyhow!("error={}", e))), + _ => Err(ApiError::internal(anyhow!("error={}", e))), + }, } } From 2803fa82ff112be77180d520cc0a2f6535435fbf Mon Sep 17 00:00:00 2001 From: nbei Date: Wed, 25 Feb 2026 14:48:20 +0800 Subject: [PATCH 7/7] fix leftover --- io-orbit/src/adapter.rs | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/io-orbit/src/adapter.rs b/io-orbit/src/adapter.rs index d113b1809..39b325c36 100644 --- a/io-orbit/src/adapter.rs +++ b/io-orbit/src/adapter.rs @@ -719,14 +719,21 @@ impl ObjectStoreAdapter { buf.extend_from_slice(&chunk); } - self.to_store() + // 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 - .map_err(IoOrbitError::from)?; - Ok(()) + { + Ok(_) => Ok(()), + Err(object_store::Error::AlreadyExists { .. }) => Ok(()), + Err(e) => Err(IoOrbitError::from(e).into()), + } } }