update code

Author: luoluoyuyu

.gitignore

@@ -4,6 +4,7 @@
 **/.vscode/
 **/.venv/**
 **/dist/**
+logs/**
 
 **.wasm
 

Cargo.lock

@@ -51,6 +51,7 @@ parking_lot = "0.12"
 arrow-array = "52"
 arrow-ipc = "52"
 arrow-schema = "52"
+proctitle = "0.1"
 
 [features]
 default = ["incremental-cache", "python"]

Cargo.toml

@@ -21,38 +21,26 @@ PACKAGE_DIR := packages
 PYTHON_WASM_PATH := python/functionstream-runtime/target/functionstream-python-runtime.wasm
 PYTHON_WASM_NAME := functionstream-python-runtime.wasm
 
-.PHONY: help clean clean-dist build build-full build-lite package package-full package-lite package-all test install docker-build docker-up fmt fmt-check
+.PHONY: help clean clean-dist build build-full build-lite package package-full package-lite package-all test install
 
 help:
 	@echo "Function Stream Build System"
 	@echo ""
 	@echo "Available targets:"
-	@echo "  fmt            - Format code with cargo fmt"
-	@echo "  fmt-check      - Check code format without modifying"
 	@echo "  build          - Build full version (debug)"
 	@echo "  build-full     - Build full release version"
 	@echo "  build-lite     - Build lite release version (no Python)"
 	@echo "  package-full   - Build and package full version (.zip and .tar.gz)"
 	@echo "  package-lite   - Build and package lite version (.zip and .tar.gz)"
 	@echo "  package-all    - Build and package both versions"
-	@echo "  docker-build   - Run package-full then docker-compose build"
-	@echo "  docker-up     - Run package-full then docker-compose up -d"
-	@echo "  test          - Run tests"
+	@echo "  test           - Run tests"
 	@echo "  clean          - Clean all build artifacts (cargo + dist)"
 	@echo "  clean-dist     - Clean distribution directory only"
 	@echo ""
 	@echo "Version: $(VERSION)"
 	@echo "Architecture: $(ARCH)"
 	@echo "OS: $(OS)"
 
-fmt:
-	@echo "Formatting code..."
-	cargo fmt --all
-
-fmt-check:
-	@echo "Checking code format..."
-	cargo fmt --all -- --check
-
 clean:
 	@echo "Cleaning build artifacts..."
 	cargo clean
@@ -232,14 +220,6 @@ package-all: clean-dist package-full package-lite
 	@echo "All packages created:"
 	@ls -lh $(DIST_BASE)/$(PACKAGE_DIR)/
 
-docker-build: package-full
-	@echo "Building Docker image (using dist/packages/function-stream-$(VERSION).zip)..."
-	docker-compose build
-
-docker-up: package-full
-	@echo "Building and starting containers..."
-	docker-compose up -d --build
-
 test:
 	cargo test
 

Makefile

@@ -22,7 +22,11 @@ pub mod wasm_config;
 pub use global_config::GlobalConfig;
 pub use loader::load_global_config;
 pub use log_config::LogConfig;
+#[allow(unused_imports)]
 pub use paths::{
-    find_config_file, find_or_create_conf_dir, find_or_create_data_dir, find_or_create_logs_dir,
+    ENV_CONF, ENV_HOME, find_config_file, get_app_log_path, get_conf_dir, get_data_dir,
+    get_log_path, get_logs_dir, get_project_root, get_python_cache_dir, get_python_cwasm_path,
+    get_python_wasm_path, get_state_dir, get_state_dir_for_base, get_task_dir, get_wasm_cache_dir,
+    resolve_path,
 };
 pub use python_config::PythonConfig;

src/config/mod.rs

@@ -10,94 +10,136 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::env;
 use std::fs;
-use std::path::{Path, PathBuf};
+use std::path::PathBuf;
+use std::sync::OnceLock;
 
-fn get_exe_dir() -> Option<PathBuf> {
-    std::env::current_exe().ok()?.parent().map(PathBuf::from)
+pub const ENV_HOME: &str = "FUNCTION_STREAM_HOME";
+pub const ENV_CONF: &str = "FUNCTION_STREAM_CONF";
+
+static PROJECT_ROOT: OnceLock<PathBuf> = OnceLock::new();
+
+pub fn get_project_root() -> &'static PathBuf {
+    PROJECT_ROOT
+        .get_or_init(|| resolve_project_root().expect("CRITICAL: Failed to resolve project root"))
 }
 
-fn search_paths<F>(mut check: F) -> Option<PathBuf>
-where
-    F: FnMut(&Path) -> bool,
-{
-    let cwd = PathBuf::from(".");
-    if check(cwd.as_path()) {
-        return Some(cwd);
+fn resolve_project_root() -> std::io::Result<PathBuf> {
+    if let Ok(home) = env::var(ENV_HOME) {
+        let path = PathBuf::from(&home);
+        return path.canonicalize().or(Ok(path));
     }
 
-    if let Some(exe_dir) = get_exe_dir() {
-        if check(exe_dir.as_path()) {
-            return Some(exe_dir);
-        }
+    if let Ok(manifest_dir) = env::var("CARGO_MANIFEST_DIR") {
+        return Ok(PathBuf::from(manifest_dir));
+    }
 
-        if let Some(parent) = exe_dir.parent() {
-            if check(parent) {
-                return Some(parent.to_path_buf());
-            }
+    if let Ok(exe_path) = env::current_exe() {
+        let mut path = exe_path;
+        path.pop();
+        if path.file_name().map_or(false, |n| n == "bin") {
+            path.pop();
         }
+        return Ok(path);
     }
 
-    None
+    env::current_dir()
 }
 
-pub fn find_config_file(config_name: &str) -> Option<PathBuf> {
-    let config_path = PathBuf::from(config_name);
-    if config_path.exists() {
-        return Some(config_path);
+pub fn resolve_path(input_path: &str) -> PathBuf {
+    let path = PathBuf::from(input_path);
+    if path.is_absolute() {
+        path
+    } else {
+        get_project_root().join(path)
     }
+}
 
-    search_paths(|base| base.join(config_name).exists())
-        .or_else(|| {
-            get_exe_dir().and_then(|exe_dir| {
-                exe_dir.parent().and_then(|parent| {
-                    let path = parent.join("conf").join(config_name);
-                    path.exists().then_some(path)
-                })
-            })
-        })
-        .or_else(|| {
-            std::env::var("FUNCTION_STREAM_CONFIG")
-                .ok()
-                .map(PathBuf::from)
-                .filter(|p| p.exists())
-        })
-}
-
-fn find_or_create_dir(dir_name: &str) -> std::io::Result<PathBuf> {
-    let cwd_dir = PathBuf::from(dir_name);
-    if cwd_dir.exists() {
-        return Ok(cwd_dir);
-    }
+fn to_absolute_path(input_path: &str) -> PathBuf {
+    resolve_path(input_path)
+}
 
-    if let Some(exe_dir) = get_exe_dir() {
-        if let Some(parent) = exe_dir.parent() {
-            let dir = parent.join(dir_name);
-            if dir.exists() {
-                return Ok(dir);
+pub fn find_config_file(config_name: &str) -> Option<PathBuf> {
+    if let Ok(conf_env) = env::var(ENV_CONF) {
+        let path = to_absolute_path(&conf_env);
+        if path.is_file() {
+            return Some(path);
+        }
+        if path.is_dir() {
+            let full = path.join(config_name);
+            if full.exists() {
+                return Some(full);
             }
-            fs::create_dir_all(&dir)?;
-            return Ok(dir);
         }
+    }
+
+    let search_paths = vec![
+        get_conf_dir().join(config_name),
+        get_project_root().join(config_name),
+    ];
 
-        let dir = exe_dir.join(dir_name);
-        if dir.exists() {
-            return Ok(dir);
+    for path in search_paths {
+        if path.exists() {
+            return Some(path.canonicalize().unwrap_or(path));
         }
     }
 
-    fs::create_dir_all(&cwd_dir)?;
-    Ok(cwd_dir)
+    None
+}
+
+fn get_or_create_sub_dir(name: &str) -> PathBuf {
+    let dir = get_project_root().join(name);
+    if !dir.exists() {
+        let _ = fs::create_dir_all(&dir);
+    }
+    dir
+}
+
+pub fn get_data_dir() -> PathBuf {
+    get_or_create_sub_dir("data")
+}
+
+pub fn get_logs_dir() -> PathBuf {
+    get_or_create_sub_dir("logs")
+}
+
+pub fn get_conf_dir() -> PathBuf {
+    get_or_create_sub_dir("conf")
+}
+
+pub fn get_task_dir() -> PathBuf {
+    get_or_create_sub_dir("data/task")
+}
+
+pub fn get_state_dir() -> PathBuf {
+    get_or_create_sub_dir("data/state")
+}
+
+pub fn get_state_dir_for_base(base: &str) -> PathBuf {
+    resolve_path(base).join("state")
+}
+
+pub fn get_app_log_path() -> PathBuf {
+    get_logs_dir().join("app.log")
+}
+
+pub fn get_log_path(relative: &str) -> PathBuf {
+    get_logs_dir().join(relative)
+}
+
+pub fn get_wasm_cache_dir() -> PathBuf {
+    get_or_create_sub_dir("data/cache/wasm-incremental")
 }
 
-pub fn find_or_create_data_dir() -> std::io::Result<PathBuf> {
-    find_or_create_dir("data")
+pub fn get_python_cache_dir() -> PathBuf {
+    get_or_create_sub_dir("data/cache/python-runner")
 }
 
-pub fn find_or_create_conf_dir() -> std::io::Result<PathBuf> {
-    find_or_create_dir("conf")
+pub fn get_python_wasm_path() -> PathBuf {
+    get_python_cache_dir().join("functionstream-python-runtime.wasm")
 }
 
-pub fn find_or_create_logs_dir() -> std::io::Result<PathBuf> {
-    find_or_create_dir("logs")
+pub fn get_python_cwasm_path() -> PathBuf {
+    get_python_cache_dir().join("functionstream-python-runtime.cwasm")
 }

src/config/paths.rs

@@ -13,24 +13,19 @@
 use serde::{Deserialize, Serialize};
 use std::path::PathBuf;
 
-/// Default directory for Python runtime cache (wasm and cwasm files)
-pub const DEFAULT_PYTHON_CACHE_DIR: &str = "data/cache/python-runner";
-
-/// Default Python WASM filename
 pub const DEFAULT_PYTHON_WASM_FILENAME: &str = "functionstream-python-runtime.wasm";
-
-/// Default Python compiled WASM cache filename
 pub const DEFAULT_PYTHON_CWASM_FILENAME: &str = "functionstream-python-runtime.cwasm";
 
 fn default_python_wasm_path() -> String {
-    format!(
-        "{}/{}",
-        DEFAULT_PYTHON_CACHE_DIR, DEFAULT_PYTHON_WASM_FILENAME
-    )
+    super::paths::get_python_wasm_path()
+        .to_string_lossy()
+        .to_string()
 }
 
 fn default_python_cache_dir() -> String {
-    DEFAULT_PYTHON_CACHE_DIR.to_string()
+    super::paths::get_python_cache_dir()
+        .to_string_lossy()
+        .to_string()
 }
 
 fn default_true() -> bool {
@@ -39,19 +34,12 @@ fn default_true() -> bool {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct PythonConfig {
-    /// Path to the Python WASM file
-    /// Default: data/cache/python-runner/functionstream-python-runtime.wasm
     #[serde(default = "default_python_wasm_path")]
     pub wasm_path: String,
 
-    /// Cache directory for precompiled components
-    /// Default: data/cache/python-runner
     #[serde(default = "default_python_cache_dir")]
     pub cache_dir: String,
 
-    /// Enable component caching
-    /// If true, precompiled components will be cached to speed up subsequent loads
-    /// Default: true
     #[serde(default = "default_true")]
     pub enable_cache: bool,
 }
@@ -67,18 +55,15 @@ impl Default for PythonConfig {
 }
 
 impl PythonConfig {
-    /// Get the WASM file path as PathBuf
     pub fn wasm_path_buf(&self) -> PathBuf {
-        PathBuf::from(&self.wasm_path)
+        super::paths::resolve_path(&self.wasm_path)
     }
 
-    /// Get the cache directory as PathBuf
     pub fn cache_dir_buf(&self) -> PathBuf {
-        PathBuf::from(&self.cache_dir)
+        super::paths::resolve_path(&self.cache_dir)
     }
 
-    /// Get the precompiled component cache file path (cwasm)
     pub fn cwasm_cache_path(&self) -> PathBuf {
-        self.cache_dir_buf().join(DEFAULT_PYTHON_CWASM_FILENAME)
+        super::paths::resolve_path(&self.cache_dir).join(DEFAULT_PYTHON_CWASM_FILENAME)
     }
 }

src/config/python_config.rs

@@ -13,7 +13,9 @@
 use serde::{Deserialize, Serialize};
 
 fn default_wasm_cache_dir() -> String {
-    ".cache/wasm-incremental".to_string()
+    crate::config::paths::get_wasm_cache_dir()
+        .to_string_lossy()
+        .to_string()
 }
 
 fn default_true() -> bool {