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..a44f04223 100644 --- a/ceres/src/api_service/mod.rs +++ b/ceres/src/api_service/mod.rs @@ -122,10 +122,15 @@ 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/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/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..17cbd9b51 100644 --- a/common/src/errors.rs +++ b/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 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/common/src/utils.rs b/common/src/utils.rs index c7b16bb8f..028f9d3e5 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() @@ -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 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/adapter.rs b/io-orbit/src/adapter.rs index 876a5e7de..39b325c36 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,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()), + } + } } 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..e134ae0e4 --- /dev/null +++ b/io-orbit/src/bin/migrate_local_to_s3.rs @@ -0,0 +1,308 @@ +//! 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, 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}; + +#[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); + // 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 { + 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(()) + })); + + // 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 { + t.await + .map_err(|e| MegaError::Other(format!("migration task panicked: {e}")))??; + } + + info!("offline migration completed"); + Ok(()) +} diff --git a/io-orbit/src/error.rs b/io-orbit/src/error.rs index 783c912a6..a468a7301 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,15 @@ 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/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/git_service.rs b/jupiter/src/service/git_service.rs index 23db8a919..4a089a1a7 100644 --- a/jupiter/src/service/git_service.rs +++ b/jupiter/src/service/git_service.rs @@ -1,5 +1,5 @@ use bytes::Bytes; -use common::errors::MegaError; +use common::{errors::MegaError, utils::is_full_hex_object_id}; use futures::StreamExt; use git_internal::internal::object::blob::Blob; use io_orbit::{ @@ -83,6 +83,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 +105,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/service/mono_service.rs b/jupiter/src/service/mono_service.rs index 537287354..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?) } @@ -154,6 +152,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 +163,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 } } diff --git a/jupiter/src/storage/mod.rs b/jupiter/src/storage/mod.rs index 7b8688456..0ce657daa 100644 --- a/jupiter/src/storage/mod.rs +++ b/jupiter/src/storage/mod.rs @@ -270,6 +270,62 @@ 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() => { + let friendly = format!( + "[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) => { + 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!( + "[obj_meta_lookup_failed] 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() } diff --git a/mono/src/api/api_router.rs b/mono/src/api/api_router.rs index b2a2ecf7c..81f7a3933 100644 --- a/mono/src/api/api_router.rs +++ b/mono/src/api/api_router.rs @@ -1,4 +1,4 @@ -use anyhow::Result; +use anyhow::{Result, anyhow}; use axum::{ Json, body::Body, @@ -7,7 +7,7 @@ use axum::{ routing::get, }; use ceres::{api_service::ApiHandler, model::git::TreeQuery}; -use http::StatusCode; +use common::errors::MegaError; use utoipa_axum::{router::OpenApiRouter, routes}; use crate::{ @@ -78,12 +78,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) => match e { + MegaError::ObjStorageNotFound(_) => Err(ApiError::not_found(anyhow!("error={}", e))), + _ => Err(ApiError::internal(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!(