fix(scorpio): stabilize Antares mounts, fix dicfuse parent.children, and add configurable HTTP bind address (#1809)

* fix(scorpio): stabilize Antares mounts, fix dicfuse parent.children, and add configurable HTTP bind address

- Stabilize Antares mounts and cleanup on shutdown
- Ensure parent.children updated in upsert_inode
- Add --http-addr CLI argument for configurable HTTP bind address

Signed-off-by: jerry609 <1772030600@qq.com>

* Fix AWS SDK BehaviorVersion deprecation

---------

Signed-off-by: jerry609 <1772030600@qq.com>

Author: jerry609

docker/dev-image/.env.example

@@ -26,6 +26,10 @@ SCORPIO_LFS_URL=http://host.docker.internal:8000
 # Scorpio HTTP port
 SCORPIO_PORT=2725
 
+# Optional: override Scorpio bind address inside container (newer scorpio supports `--http-addr`)
+# If you set this, ensure your port mappings / SCORPIO_API_BASE_URL match.
+# SCORPIO_HTTP_ADDR=0.0.0.0:2725
+
 # Store path for Scorpio data (inside container)
 SCORPIO_STORE_PATH=/data/scorpio/store
 

docker/dev-image/entrypoint.sh

@@ -249,7 +249,16 @@ start_scorpio() {
     echo ""
 
     log_info "Starting scorpio..."
-    exec scorpio -c "$SCORPIO_CONFIG" "$@"
+    local extra_args=()
+    # Optional: bind HTTP server to a custom address/port (supported by newer scorpio binaries).
+    # If user already passed --http-addr in args, do not add a duplicate flag.
+    if [ -n "${SCORPIO_HTTP_ADDR:-}" ]; then
+        case " $* " in
+            *" --http-addr "*|*" --http-addr="*) ;;
+            *) extra_args+=(--http-addr "${SCORPIO_HTTP_ADDR}") ;;
+        esac
+    fi
+    exec scorpio -c "$SCORPIO_CONFIG" "${extra_args[@]}" "$@"
 }
 
 start_orion_worker() {
@@ -274,10 +283,28 @@ start_orion_worker() {
         check_fuse
         generate_scorpio_config
 
-        # Ensure the worker talks to the embedded Scorpio by default.
-        export SCORPIO_API_BASE_URL="http://127.0.0.1:2725"
+        local extra_args=()
+        local http_addr="${SCORPIO_HTTP_ADDR:-0.0.0.0:2725}"
+        # NOTE: Simple IPv4-style parsing. For IPv6 bind addresses, set SCORPIO_API_BASE_URL manually.
+        local http_port="${http_addr##*:}"
+        case "$http_port" in
+            ''|*[!0-9]*)
+                http_port="2725"
+                ;;
+        esac
+
+        # Ensure the worker talks to the embedded Scorpio by default (respecting http port override).
+        export SCORPIO_API_BASE_URL="http://127.0.0.1:${http_port}"
+
+        # Optional: pass through --http-addr if provided (and not already present in args).
+        if [ -n "${SCORPIO_HTTP_ADDR:-}" ]; then
+            case " $* " in
+                *" --http-addr "*|*" --http-addr="*) ;;
+                *) extra_args+=(--http-addr "${SCORPIO_HTTP_ADDR}") ;;
+            esac
+        fi
 
-        scorpio -c "$SCORPIO_CONFIG" &
+        scorpio -c "$SCORPIO_CONFIG" "${extra_args[@]}" &
         local scorpio_pid=$!
 
         cleanup() {
@@ -289,7 +316,7 @@ start_orion_worker() {
 
         trap cleanup EXIT INT TERM
 
-        wait_for_service "127.0.0.1" "2725" 60 || exit 1
+        wait_for_service "127.0.0.1" "${http_port}" 60 || exit 1
     else
         log_warn "ORION_WORKER_START_SCORPIO disabled; ensure Scorpio mountpoints are accessible to this container."
     fi

orion-server/src/log/store/s3_log_store.rs

@@ -24,7 +24,7 @@ impl S3LogStore {
     ) -> Self {
         let region = Region::new(region_name.to_string());
 
-        let shared_config = aws_config::defaults(BehaviorVersion::v2025_08_07())
+        let shared_config = aws_config::defaults(BehaviorVersion::latest())
             .region(region.clone())
             .credentials_provider(Credentials::new(
                 access_key_id,

scorpio/src/antares/fuse.rs

@@ -43,8 +43,13 @@ impl AntaresFuse {
 
     /// Compose the union filesystem instance.
     pub async fn build_overlay(&self) -> std::io::Result<OverlayFs> {
-        // Build lower layers: optional CL, then a passthrough over the upper dir as a fallback lower.
+        // Build lower layers:
+        // - Prefer Dicfuse as the primary lower layer (read-only monorepo view).
+        // - Optional CL dir is best-effort and may be empty; keep it as an additional
+        //   layer so it never masks the Dicfuse view.
         let mut lower_layers: Vec<Arc<dyn Layer>> = Vec::new();
+        lower_layers.push(self.dic.clone() as Arc<dyn Layer>);
+
         if let Some(cl_dir) = &self.cl_dir {
             let cl_layer = new_passthroughfs_layer(PassthroughArgs {
                 root_dir: cl_dir,
@@ -54,8 +59,6 @@ impl AntaresFuse {
             lower_layers.push(Arc::new(cl_layer) as Arc<dyn Layer>);
         }
 
-        lower_layers.push(self.dic.clone() as Arc<dyn Layer>);
-
         // Upper layer mirrors upper_dir to keep writes separated from lower layers.
         let upper_layer: Arc<dyn Layer> = Arc::new(
             new_passthroughfs_layer(PassthroughArgs {

scorpio/src/daemon/antares.rs

@@ -542,8 +542,35 @@ impl AntaresServiceImpl {
     /// - Add progress callback for long-running initialization
     /// - Consider lazy loading for very large subdirectory mounts
     async fn get_or_create_dicfuse(&self, path: &str) -> Result<Arc<Dicfuse>, ServiceError> {
-        // For root path, use the shared global instance
+        const INIT_TIMEOUT_SECS: u64 = 120;
+
+        // For root path, use the shared global instance (but ensure it's initialized first).
         if path.is_empty() || path == "/" {
+            tracing::info!(
+                "Waiting for shared Dicfuse instance to initialize for path: / (timeout: {}s)",
+                INIT_TIMEOUT_SECS
+            );
+            match tokio::time::timeout(
+                Duration::from_secs(INIT_TIMEOUT_SECS),
+                self.dicfuse.store.wait_for_ready(),
+            )
+            .await
+            {
+                Ok(_) => {
+                    tracing::info!("Shared Dicfuse initialized successfully for path: /");
+                }
+                Err(_) => {
+                    tracing::error!(
+                        "Shared Dicfuse initialization timed out for path: / after {}s",
+                        INIT_TIMEOUT_SECS
+                    );
+                    return Err(ServiceError::FuseFailure(format!(
+                        "Dicfuse initialization timed out for path '/' after {}s. \
+                         Check network connectivity to the monorepo server.",
+                        INIT_TIMEOUT_SECS
+                    )));
+                }
+            }
             return Ok(self.dicfuse.clone());
         }
 
@@ -568,7 +595,6 @@ impl AntaresServiceImpl {
         // CRITICAL: Wait for the Dicfuse directory tree to be fully loaded before
         // returning. Without this, FUSE mount may fail because the root inode
         // is not set up yet when import_arc hasn't completed.
-        const INIT_TIMEOUT_SECS: u64 = 120;
         // TODO(dicfuse-antares-integration): If many concurrent requests initialize DIFFERENT
         // base paths, we may enqueue a large number of concurrent warmups (network + memory).
         // Consider adding a global semaphore/queue to cap concurrent initializations.

scorpio/src/daemon/mod.rs

@@ -1,3 +1,4 @@
+use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::sync::Arc;
 
@@ -11,7 +12,7 @@ use axum::Router;
 use dashmap::DashMap;
 use git_internal::hash::ObjectHash;
 use serde::{Deserialize, Serialize};
-use tokio::sync::Mutex;
+use tokio::sync::{oneshot, Mutex};
 use uuid::Uuid;
 pub mod antares;
 //mod git;
@@ -105,7 +106,12 @@ struct ScoState {
     tasks: Arc<DashMap<String, MountStatus>>, // Thread-safe storage for async mount tasks
 }
 #[allow(unused)]
-pub async fn daemon_main(fuse: Arc<MegaFuse>, manager: ScorpioManager) {
+pub async fn daemon_main(
+    fuse: Arc<MegaFuse>,
+    manager: ScorpioManager,
+    shutdown_rx: oneshot::Receiver<()>,
+    bind_addr: SocketAddr,
+) {
     let inner = ScoState {
         fuse,
         manager: Arc::new(Mutex::new(manager)),
@@ -125,12 +131,29 @@ pub async fn daemon_main(fuse: Arc<MegaFuse>, manager: ScorpioManager) {
 
     // Antares route - create service with new Dicfuse instance
     let antares_service = Arc::new(antares::AntaresServiceImpl::new(None).await);
+    let antares_service_for_shutdown = antares_service.clone();
     let antares_daemon = antares::AntaresDaemon::new(antares_service);
     let antares_router = antares_daemon.router();
     let app = app.nest("/antares", antares_router);
 
-    let listener = tokio::net::TcpListener::bind("0.0.0.0:2725").await.unwrap();
-    axum::serve(listener, app).await.unwrap()
+    let listener = tokio::net::TcpListener::bind(bind_addr).await.unwrap();
+    axum::serve(listener, app)
+        .with_graceful_shutdown(async move {
+            let _ = shutdown_rx.await;
+            tracing::info!("HTTP server shutdown requested; running Antares shutdown cleanup");
+            match tokio::time::timeout(
+                std::time::Duration::from_secs(15),
+                antares_service_for_shutdown.shutdown_cleanup_impl(),
+            )
+            .await
+            {
+                Ok(Ok(())) => tracing::info!("Antares shutdown cleanup completed"),
+                Ok(Err(e)) => tracing::warn!("Antares shutdown cleanup failed: {:?}", e),
+                Err(_) => tracing::warn!("Antares shutdown cleanup timed out"),
+            }
+        })
+        .await
+        .unwrap()
 }
 
 /// Asynchronous mount handler for clients.

scorpio/src/dicfuse/store.rs

@@ -1068,6 +1068,14 @@ impl DictionaryStore {
         };
 
         if let Some(inode) = existing {
+            // Ensure parent.children contains this inode (fix inconsistent state from DB recovery).
+            // This handles cases where radix_trie has the path->inode mapping but parent.children
+            // list is missing the inode entry.
+            if let Ok(parent_item) = self.persistent_path_store.get_item(parent) {
+                if !parent_item.get_children().contains(&inode) {
+                    let _ = self.persistent_path_store.append_child(parent, inode);
+                }
+            }
             return Ok(inode);
         }
 

scorpio/src/main.rs

@@ -5,8 +5,11 @@ use scorpio::fuse::MegaFuse;
 use scorpio::manager::{fetch::CheckHash, ScorpioManager};
 use scorpio::server::mount_filesystem;
 use scorpio::util::config;
+use std::net::SocketAddr;
 use std::{ffi::OsStr, sync::Arc};
+#[cfg(not(unix))]
 use tokio::signal;
+use tokio::sync::oneshot;
 
 /// Command line arguments for the application
 #[derive(Parser, Debug)]
@@ -15,6 +18,10 @@ struct Args {
     /// Path to the configuration file
     #[arg(short, long, default_value = "scorpio.toml")]
     config_path: String,
+
+    /// HTTP bind address for the daemon (Antares API lives under /antares/*)
+    #[arg(long, default_value = "0.0.0.0:2725")]
+    http_addr: SocketAddr,
 }
 
 #[tokio::main]
@@ -44,19 +51,61 @@ async fn main() {
     let lgfs = LoggingFileSystem::new(fuse_interface.clone());
     let mut mount_handle = mount_filesystem(lgfs, mountpoint).await;
 
-    let handle = &mut mount_handle;
+    print!("server running...");
 
-    // spawn the server running function.
-    tokio::spawn(daemon_main(Arc::new(fuse_interface), manager));
+    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
+    let mut daemon_task = tokio::spawn(daemon_main(
+        Arc::new(fuse_interface),
+        manager,
+        shutdown_rx,
+        args.http_addr,
+    ));
 
-    print!("server running...");
+    let mut mount_finished = false;
     tokio::select! {
-        res = handle => res.unwrap(),
-        _ = signal::ctrl_c() => {
+        res = &mut mount_handle => {
+            mount_finished = true;
+            if let Err(e) = res {
+                eprintln!("FUSE session ended with error: {e:?}");
+            }
+        }
+        _ = shutdown_signal() => {
+            // fallthrough to shutdown sequence below
+        }
+    }
 
-            println!("unmount....");
-            mount_handle.unmount().await.unwrap();
+    // Stop HTTP server first (this triggers Antares shutdown cleanup), then unmount the main workspace FS.
+    let _ = shutdown_tx.send(());
+    match tokio::time::timeout(std::time::Duration::from_secs(20), &mut daemon_task).await {
+        Ok(Ok(_)) => {}
+        Ok(Err(e)) => eprintln!("HTTP daemon task join failed: {e}"),
+        Err(_) => {
+            eprintln!("HTTP daemon shutdown timed out; aborting task");
+            daemon_task.abort();
+        }
+    }
+
+    if !mount_finished {
+        println!("unmount....");
+        let _ = mount_handle.unmount().await;
+    }
+}
 
+async fn shutdown_signal() {
+    #[cfg(unix)]
+    {
+        let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
+            .expect("failed to install SIGTERM handler");
+        let mut sigint = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())
+            .expect("failed to install SIGINT handler");
+        tokio::select! {
+            _ = sigterm.recv() => {}
+            _ = sigint.recv() => {}
         }
     }
+
+    #[cfg(not(unix))]
+    {
+        let _ = signal::ctrl_c().await;
+    }
 }