docs/remote-config.md (1)

36-40:Add language specifiers to fenced code blocks.

The static analysis correctly identifies that fenced code blocks should have language specifiers for better syntax highlighting.

Apply these diffs to add language specifiers:

For line 36:

-```+```http GET /pipelines Authorization: Bearer <token>  # 如果提供了token Content-Type: application/json

For line 176:```diff-```+```log 2024-01-15T10:30:00Z INFO Starting remote configuration manager 2024-01-15T10:30:00Z INFO Polling interval: 30 seconds 2024-01-15T10:30:00Z INFO API endpoint: http://api.example.com/pipelines 2024-01-15T10:30:30Z INFO Configuration changed, updating pipelines (version: v1.0.1) 2024-01-15T10:30:30Z INFO Starting new pipeline 'Data Processing Pipeline' 2024-01-15T10:30:30Z INFO Pipeline 'Data Processing Pipeline' started successfully

Also applies to: 176-183</blockquote></details><details><summary>crates/arkflow-core/src/remote_config.rs (3)</summary><blockquote>`210-214`: **Consider graceful task termination instead of abort.**Using `abort()` on task handles forcefully terminates tasks without cleanup. Consider awaiting cancellation instead.Instead of immediately aborting, give tasks a chance to shut down gracefully:```diff // Stop existing stream existing.cancellation_token.cancel(); if let Some(handle) = &existing.handle {-    let _ = handle.abort();+    // Give the task a chance to shut down gracefully+    tokio::select! {+        _ = tokio::time::sleep(Duration::from_secs(5)) => {+            // Timeout - force abort+            handle.abort();+        }+        _ = handle => {+            // Task completed gracefully+        }+    } }

Also applies to: 228-231, 237-241, 257-260

---

248-248:Fix typo in comment.

-        // Remove stream that are no longer in the configuration+        // Remove streams that are no longer in the configuration

---

168-186:Add retry logic for transient network failures.

Network requests can fail due to transient issues. Consider adding retry logic with exponential backoff.

Consider using a retry mechanism for better resilience:

use tokio_retry::{Retry, strategy::ExponentialBackoff};asyncfnfetch_remote_config_with_retry(&self) ->Result<RemoteConfigResponse,Box<dyn std::error::Error>>{let retry_strategy =ExponentialBackoff::from_millis(100).max_delay(Duration::from_secs(10)).take(3);Retry::spawn(retry_strategy, ||async{self.fetch_remote_config().await}).await}

This would require adding thetokio-retry crate as a dependency.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

crates/arkflow-core/src/cli/mod.rs (1)

131-131:Avoidunwrap() onself.config

This can panic ifparse() returned early under--validate, leavingself.config asNone. Prefer returning a proper error.

Apply this diff:

-            let config = self.config.clone().unwrap();+            let config = self+                .config+                .clone()+                .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::Other, "Configuration not found"))?;

crates/arkflow-core/src/cli/mod.rs (7)

107-111:Don’t exit the process inside library-style parsing; propagate errors instead

Callingprocess::exit(1) here prevents callers from handling the error and makes testing harder. Return the error and let the top-level binary decide.

-                Err(e) => {-                    println!("Failed to load configuration file: {}", e);-                    process::exit(1);-                }+                Err(e) => {+                    eprintln!("Failed to load configuration file: {}", e);+                    return Err(Box::new(e));+                }

---

113-121:--validate early return can leaveself.config unset; set it before returning

If callers invokerun() afterparse(), this leads to a panic at theunwrap(). To be safe, setself.config before the early return.

-            // If you just verify the configuration, exit it-            if matches.get_flag("validate") {-                info!("The config is validated.");-                return Ok(());-            }--            self.config = Some(config);+            // If you just verify the configuration, exit it+            if matches.get_flag("validate") {+                self.config = Some(config);+                info!("The config is validated.");+                return Ok(());+            }++            self.config = Some(config);

---

58-80:Consider making--token optional and allowing ENV override for flexibility

RemoteConfigManager::new takesOption<String>, but the CLI requires--token. Making it optional (and supporting an env var) lets users access unauthenticated endpoints or configure tokens via environment.

-                .arg(-                    Arg::new("token")-                        .long("token")-                        .value_name("TOKEN")-                        .help("Authentication token for remote configuration API")-                        .required( true),-                ))+                .arg(+                    Arg::new("token")+                        .long("token")+                        .value_name("TOKEN")+                        .help("Authentication token for remote configuration API")+                        .env("ARKFLOW_REMOTE_TOKEN")+                ))

---

91-96:Avoidexpect() for URL; propagate a user-friendly error instead

Even though clap enforcesrequired(true), prefer consistent error handling by returning an error rather than panicking.

-            let remote_url = remote-                .get_one::<String>("url")-                .expect("Remote configuration URL not found");+            let remote_url = remote+                .get_one::<String>("url")+                .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidInput, "Remote configuration URL not found"))?;

---

20-21:Unix-only signal handling; consider cross-platform support

Importingtokio::signal::unix limits builds to Unix targets. If Windows support is desired, add actrl_c fallback with conditional compilation.

Example changes:

- use tokio::signal::unix::{signal, SignalKind};+ #[cfg(unix)]+ use tokio::signal::unix::{signal, SignalKind};+ #[cfg(not(unix))]+ use tokio::signal;

And inrun():

#[cfg(unix)]// existing SIGINT/SIGTERM handling#[cfg(not(unix))]tokio::spawn({let token = token.clone();asyncmove{if signal::ctrl_c().await.is_ok(){info!("Received Ctrl+C, exiting...");}        token.cancel();}});

---

156-224:Nit: remove explicit-> () and avoid.pretty() with JSON format

Minor polish:

• Returning() explicitly is redundant in Rust.
• When using.json(),.pretty() is ignored; drop it to reduce confusion.

-pub(crate) fn init_logging(config: &EngineConfig) -> () {+pub(crate) fn init_logging(config: &EngineConfig) {@@-                    LogFormat::JSON => {+                    LogFormat::JSON => {                         let subscriber = subscriber_builder                             .with_writer(std::sync::Mutex::new(file))-                            .pretty()                             .json()                             .finish();                         tracing::subscriber::set_global_default(subscriber)                             .expect("You can't set a global default log subscriber");                     }@@     match config.logging.format {         LogFormat::JSON => {-            let subscriber = subscriber_builder.pretty().json().finish();+            let subscriber = subscriber_builder.json().finish();             tracing::subscriber::set_global_default(subscriber)                 .expect("You can't set a global default log subscriber");         }

---

43-55:Nit: help text wording improvements

Tighten phrasing for clarity and correctness.

-            .about("High-performance Rust stream processing engine, providing powerful data stream processing capabilities, supporting multiple input/output sources and processors")+            .about("High-performance Rust stream processing engine with multiple inputs/outputs and processors")@@-                    .help("Specify the profile path")+                    .help("Path to the configuration file")@@-                    .help("Only the profile is verified, not the engine is started")+                    .help("Validate the configuration file without starting the engine")

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.