feat(orion): fan-out image publish and public presign endpoint

Allow build-custom-image to upload/register across envs, avoid Start
Runner hangs behind the update lock, and sign Orion image URLs with an
optional public RustFS host so out-of-cluster schedulers can download.

Author: benjamin-747

clients/orion-scheduler-client/src/http_client.rs

@@ -59,16 +59,24 @@ impl OrionSchedulerHttpClient {
             "Starting runner via scheduler: server_ws={}",
             payload.server_ws
         );
+        // Conflict checks must not block behind a multi-minute image download;
+        // scheduler returns 503 quickly when the update lock is busy. Keep a
+        // modest client budget for network + signing + lock try.
         let req = self
             .client
             .post(&url)
-            .timeout(Duration::from_secs(10))
+            .timeout(Duration::from_secs(30))
             .json(&payload);
         let res = self.auth_headers(req).send().await?;
         let status = res.status();
         let body: StartRunnerSchedulerResponse = res.json().await?;
-        // 200 OK (idempotent), 202 Accepted (provisioning), 409 Conflict
-        if status.is_success() || status.as_u16() == 202 || status.as_u16() == 409 {
+        // 200 OK (idempotent), 202 Accepted (provisioning), 409 Conflict,
+        // 503 Busy (another provision holds the update lock).
+        if status.is_success()
+            || status.as_u16() == 202
+            || status.as_u16() == 409
+            || status.as_u16() == 503
+        {
             Ok(body)
         } else {
             Err(anyhow::anyhow!(

clients/orion-scheduler-client/src/lib.rs

@@ -28,6 +28,19 @@ pub struct StartRunnerPayload {
     pub image_cpus: Option<u32>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub image_memory_mb: Option<u32>,
+    /// Catalog metadata (from mono when starting via `image_id`).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub image_name: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub image_built_at: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub toolchain_rust: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub toolchain_buck2: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub toolchain_python: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub kernel: Option<String>,
     /// When set, write `ORION_RETAIN_ANTARES_MOUNTS` into the guest `.env`.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub retain_antares_mounts: Option<bool>,

common/src/config/mod.rs

@@ -614,6 +614,13 @@ pub struct S3Config {
     pub access_key_id: String,
     pub secret_access_key: String,
     pub endpoint_url: String,
+    /// Optional endpoint used only when generating presigned URLs.
+    ///
+    /// Keep `endpoint_url` as the in-cluster address for mono PUT/GET, and set
+    /// this to a host that out-of-cluster clients (e.g. orion-scheduler) can
+    /// reach. Empty = sign with `endpoint_url` (unchanged behavior).
+    #[serde(default)]
+    pub presign_endpoint_url: String,
 }
 
 #[derive(Debug, Serialize, Deserialize, Default, Clone)]

config/config.toml

@@ -152,6 +152,15 @@ secret_access_key = ""
 # set this to the service endpoint.
 endpoint_url = "http://localhost:9000"
 
+# Optional: endpoint used only for presigned GET/PUT URLs.
+# Use when mono talks to RustFS over an in-cluster URL but external clients
+# (orion-scheduler on the host) need a public hostname. Signature includes
+# Host, so this must match the URL clients will open. Empty = use endpoint_url.
+# Example:
+#   endpoint_url = "http://rustfs.mega-dev.svc.cluster.local:9000"
+#   presign_endpoint_url = "https://rustfs.xuanwu.openatom.cn"
+presign_endpoint_url = ""
+
 
 [object_storage.gcs]
 # Name of the GCS bucket

io-orbit/src/adapter.rs

@@ -41,6 +41,9 @@ pub enum UploadStrategy {
 pub struct ObjectStoreAdapter {
     /// The concrete backend store used for all object operations.
     pub store: BackendStore,
+    /// Optional S3 client used only for `signed_url` (public/presign endpoint).
+    /// When `None`, signing uses [`Self::store`].
+    pub presign_store: Option<Arc<AmazonS3>>,
     /// The upload strategy used when writing new objects.
     pub upload_strategy: UploadStrategy,
 }
@@ -165,6 +168,16 @@ impl MegaObjectStorage for ObjectStoreAdapter {
     ) -> Result<Option<String>, MegaError> {
         let path = key.to_object_store_path();
 
+        if let Some(presign) = &self.presign_store {
+            return Ok(Some(
+                presign
+                    .signed_url(method, &path, expires_in)
+                    .await
+                    .map_err(IoOrbitError::from)?
+                    .to_string(),
+            ));
+        }
+
         let url = match &self.store {
             BackendStore::S3(s3) => Some(
                 s3.signed_url(method, &path, expires_in)
@@ -1171,6 +1184,7 @@ mod tests {
         let local = Arc::new(LocalFileSystem::new_with_prefix(dir.path()).unwrap());
         let adapter = ObjectStoreAdapter {
             store: BackendStore::Local(Arc::clone(&local)),
+            presign_store: None,
             upload_strategy: UploadStrategy::SinglePut,
         };
         let key = ObjectKey {

io-orbit/src/factory.rs

@@ -4,10 +4,14 @@ use std::{
 };
 
 use common::{
-    config::{ObjectStorageBackend, ObjectStorageConfig},
+    config::{ObjectStorageBackend, ObjectStorageConfig, S3Config},
     errors::MegaError,
 };
-use object_store::{aws::AmazonS3Builder, gcp::GoogleCloudStorageBuilder, local::LocalFileSystem};
+use object_store::{
+    aws::{AmazonS3, AmazonS3Builder},
+    gcp::GoogleCloudStorageBuilder,
+    local::LocalFileSystem,
+};
 
 use crate::{
     adapter::{BackendStore, ObjectStoreAdapter, UploadStrategy},
@@ -42,6 +46,7 @@ impl MegaObjectStorageWrapper {
         let store = BackendStore::Local(Arc::new(fs));
         let adapter = Arc::new(ObjectStoreAdapter {
             store,
+            presign_store: None,
             upload_strategy: UploadStrategy::SinglePut,
         });
         MegaObjectStorageWrapper::new(adapter)
@@ -62,26 +67,58 @@ impl ObjectStorageFactory {
     }
 }
 
-/// Shared S3 / S3-compatible construction (differs only by endpoint and upload strategy).
-async fn build_s3_like(
-    cfg: &ObjectStorageConfig,
+fn build_amazon_s3(
+    s3_cfg: &S3Config,
+    endpoint: &str,
     compatible: bool,
-) -> Result<MegaObjectStorageWrapper, MegaError> {
-    let s3_cfg = cfg.s3.clone();
+) -> Result<AmazonS3, MegaError> {
     let mut builder = 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);
-    if compatible {
-        builder = builder
-            .with_endpoint(&s3_cfg.endpoint_url)
-            .with_allow_http(true)
-            .with_virtual_hosted_style_request(false);
+
+    if compatible || !endpoint.is_empty() {
+        if !endpoint.is_empty() {
+            builder = builder.with_endpoint(endpoint);
+        }
+        if compatible {
+            // S3-compatible (RustFS/MinIO): path-style; allow plain HTTP for
+            // in-cluster endpoints. HTTPS public endpoints still work with
+            // allow_http(true).
+            builder = builder
+                .with_allow_http(true)
+                .with_virtual_hosted_style_request(false);
+        } else if endpoint.starts_with("http://") {
+            builder = builder.with_allow_http(true);
+        }
     }
-    let s3 = builder
-        .build()
-        .map_err(|e| MegaError::Other(e.to_string()))?;
+
+    builder.build().map_err(|e| MegaError::Other(e.to_string()))
+}
+
+/// Shared S3 / S3-compatible construction (differs only by endpoint and upload strategy).
+async fn build_s3_like(
+    cfg: &ObjectStorageConfig,
+    compatible: bool,
+) -> Result<MegaObjectStorageWrapper, MegaError> {
+    let s3_cfg = &cfg.s3;
+    let endpoint = if compatible {
+        s3_cfg.endpoint_url.as_str()
+    } else {
+        // Real AWS: empty endpoint uses the default regional endpoint.
+        ""
+    };
+    let s3 = build_amazon_s3(s3_cfg, endpoint, compatible)?;
+
+    let presign_store = {
+        let presign_ep = s3_cfg.presign_endpoint_url.trim();
+        if presign_ep.is_empty() {
+            None
+        } else {
+            Some(Arc::new(build_amazon_s3(s3_cfg, presign_ep, compatible)?))
+        }
+    };
 
     let store = BackendStore::S3(Arc::new(s3));
     let upload_strategy = if compatible {
@@ -91,6 +128,7 @@ async fn build_s3_like(
     };
     let adapter = Arc::new(ObjectStoreAdapter {
         store,
+        presign_store,
         upload_strategy,
     });
 
@@ -106,6 +144,7 @@ async fn build_gcs(cfg: &ObjectStorageConfig) -> Result<MegaObjectStorageWrapper
     let store = BackendStore::Gcs(Arc::new(gcs));
     let adapter = Arc::new(ObjectStoreAdapter {
         store,
+        presign_store: None,
         upload_strategy: UploadStrategy::SinglePut,
     });
 
@@ -122,8 +161,103 @@ async fn build_local(cfg: &ObjectStorageConfig) -> Result<MegaObjectStorageWrapp
     let store = BackendStore::Local(Arc::new(fs));
     let adapter = Arc::new(ObjectStoreAdapter {
         store,
+        presign_store: None,
         upload_strategy: UploadStrategy::SinglePut,
     });
 
     Ok(MegaObjectStorageWrapper::new(adapter))
 }
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use common::config::{
+        GcsConfig, LocalConfig, ObjectStorageBackend, ObjectStorageConfig, S3Config,
+    };
+    use object_store::{path::Path, signer::Signer};
+    use reqwest::Method;
+
+    use super::build_amazon_s3;
+    use crate::object_storage::{ObjectKey, ObjectNamespace};
+
+    fn sample_s3_config(endpoint: &str, presign: &str) -> S3Config {
+        S3Config {
+            region: "us-east-1".into(),
+            bucket: "buck2hub-assets".into(),
+            access_key_id: "testkey".into(),
+            secret_access_key: "testsecret".into(),
+            endpoint_url: endpoint.into(),
+            presign_endpoint_url: presign.into(),
+        }
+    }
+
+    #[tokio::test]
+    async fn signed_url_uses_endpoint_url_when_presign_unset() {
+        let s3_cfg = sample_s3_config("http://rustfs.internal:9000", "");
+        let s3 = build_amazon_s3(&s3_cfg, &s3_cfg.endpoint_url, true).unwrap();
+        let path = Path::from("orion-images/abc/debian-13-buck2.qcow2");
+        let url = s3
+            .signed_url(Method::GET, &path, Duration::from_secs(60))
+            .await
+            .unwrap()
+            .to_string();
+        assert!(
+            url.starts_with("http://rustfs.internal:9000/"),
+            "expected internal host, got {url}"
+        );
+    }
+
+    #[tokio::test]
+    async fn signed_url_uses_presign_endpoint_when_set() {
+        let s3_cfg = sample_s3_config(
+            "http://rustfs.mega-dev.svc.cluster.local:9000",
+            "https://rustfs.xuanwu.openatom.cn",
+        );
+        let presign = build_amazon_s3(&s3_cfg, &s3_cfg.presign_endpoint_url, true).unwrap();
+        let path = Path::from("orion-images/abc/debian-13-buck2.qcow2");
+        let url = presign
+            .signed_url(Method::GET, &path, Duration::from_secs(60))
+            .await
+            .unwrap()
+            .to_string();
+        assert!(
+            url.starts_with("https://rustfs.xuanwu.openatom.cn/"),
+            "expected public host, got {url}"
+        );
+        assert!(
+            !url.contains("svc.cluster.local"),
+            "presign URL must not use in-cluster host: {url}"
+        );
+    }
+
+    #[tokio::test]
+    async fn factory_wires_presign_store_for_compatible() {
+        use crate::factory::ObjectStorageFactory;
+
+        let cfg = ObjectStorageConfig {
+            storage_type: ObjectStorageBackend::S3Compatible,
+            s3: sample_s3_config(
+                "http://rustfs.internal:9000",
+                "https://rustfs.example.public",
+            ),
+            gcs: GcsConfig::default(),
+            local: LocalConfig::default(),
+        };
+        let wrapper = ObjectStorageFactory::build(&cfg).await.unwrap();
+        let key = ObjectKey {
+            namespace: ObjectNamespace::OrionImage,
+            key: "deadbeef/debian-13-buck2.qcow2".into(),
+        };
+        let url = wrapper
+            .inner
+            .signed_url(&key, Method::GET, Duration::from_secs(120))
+            .await
+            .unwrap()
+            .expect("s3compatible must support presign");
+        assert!(
+            url.starts_with("https://rustfs.example.public/"),
+            "factory should sign with presign_endpoint_url, got {url}"
+        );
+    }
+}

jupiter/src/service/orion_vm_image_service.rs

@@ -42,6 +42,11 @@ impl OrionVmImageService {
         self.st.list_all().await
     }
 
+    /// Newest catalog row by `created_at` (same order as [`Self::list`]).
+    pub async fn latest(&self) -> Result<Option<orion_vm_image::Model>, MegaError> {
+        Ok(self.list().await?.into_iter().next())
+    }
+
     pub async fn get(&self, id: &str) -> Result<Option<orion_vm_image::Model>, MegaError> {
         self.st.find_by_id(id).await
     }