From b2c45bf5c16391917caccd703ac85b80c5c77cca Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Fri, 24 Oct 2025 14:47:20 +0800 Subject: Re-export subcrate `action_system` to `just_enough_vcs` --- crates/vcs_actions/src/actions/local_actions.rs | 12 +++++++++++- crates/vcs_actions/src/registry/client_registry.rs | 5 +++-- 2 files changed, 14 insertions(+), 3 deletions(-) (limited to 'crates/vcs_actions') diff --git a/crates/vcs_actions/src/actions/local_actions.rs b/crates/vcs_actions/src/actions/local_actions.rs index b11a934..55d014e 100644 --- a/crates/vcs_actions/src/actions/local_actions.rs +++ b/crates/vcs_actions/src/actions/local_actions.rs @@ -1,4 +1,6 @@ -use action_system::{action::ActionContext, action_gen}; +use std::net::SocketAddr; + +use action_system::{action::ActionContext, macros::action_gen}; use log::info; use tcp_connection::error::TcpTargetError; @@ -23,3 +25,11 @@ pub async fn hello_world_action(ctx: ActionContext, _n: ()) -> Result<(), TcpTar Ok(()) } + +#[action_gen] +pub async fn set_upstream_vault_action( + ctx: ActionContext, + upstream: SocketAddr, +) -> Result<(), TcpTargetError> { + Ok(()) +} diff --git a/crates/vcs_actions/src/registry/client_registry.rs b/crates/vcs_actions/src/registry/client_registry.rs index 5939bed..484c4f4 100644 --- a/crates/vcs_actions/src/registry/client_registry.rs +++ b/crates/vcs_actions/src/registry/client_registry.rs @@ -2,12 +2,13 @@ use action_system::{action::ActionContext, action_pool::ActionPool}; use tcp_connection::error::TcpTargetError; use crate::{ - actions::local_actions::register_hello_world_action, connection::protocol::RemoteActionInvoke, + actions::local_actions::register_set_upstream_vault_action, + connection::protocol::RemoteActionInvoke, }; fn register_actions(pool: &mut ActionPool) { // Pool register here - register_hello_world_action(pool); + register_set_upstream_vault_action(pool); } pub fn client_action_pool() -> ActionPool { -- cgit From ac71a819dd45090a2ee1054208fd027f05a8c36c Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Fri, 24 Oct 2025 18:24:04 +0800 Subject: Fix some spelling issues. --- crates/vcs_actions/src/actions/local_actions.rs | 19 ++++++++++++++++++- crates/vcs_actions/src/registry/server_registry.rs | 4 ++-- 2 files changed, 20 insertions(+), 3 deletions(-) (limited to 'crates/vcs_actions') diff --git a/crates/vcs_actions/src/actions/local_actions.rs b/crates/vcs_actions/src/actions/local_actions.rs index 55d014e..c19b8f0 100644 --- a/crates/vcs_actions/src/actions/local_actions.rs +++ b/crates/vcs_actions/src/actions/local_actions.rs @@ -29,7 +29,24 @@ pub async fn hello_world_action(ctx: ActionContext, _n: ()) -> Result<(), TcpTar #[action_gen] pub async fn set_upstream_vault_action( ctx: ActionContext, - upstream: SocketAddr, + _upstream: SocketAddr, ) -> Result<(), TcpTargetError> { + // Ensure the instance is available + let Some(instance) = ctx.instance() else { + return Err(TcpTargetError::NotFound( + "Connection Instance Lost.".to_string(), + )); + }; + + if ctx.is_local() { + // Invoke on local + // Send the message to the server + let _ = instance.lock().await.write_text("Hello World!").await; + } else if ctx.is_remote() { + // Read the message from the client + let read = instance.lock().await.read_text().await?; + info!("{}", read) + } + Ok(()) } diff --git a/crates/vcs_actions/src/registry/server_registry.rs b/crates/vcs_actions/src/registry/server_registry.rs index b449b68..3ecc103 100644 --- a/crates/vcs_actions/src/registry/server_registry.rs +++ b/crates/vcs_actions/src/registry/server_registry.rs @@ -1,9 +1,9 @@ use action_system::action_pool::ActionPool; -use crate::actions::local_actions::register_hello_world_action; +use crate::actions::local_actions::register_set_upstream_vault_action; pub fn server_action_pool() -> ActionPool { let mut pool = ActionPool::new(); - register_hello_world_action(&mut pool); + register_set_upstream_vault_action(&mut pool); pool } -- cgit From e5238eebd2bc9cfbb508d7b69b3b84708bf184f7 Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Mon, 27 Oct 2025 15:23:28 +0800 Subject: Add some debug logs --- crates/vcs_actions/src/actions/local_actions.rs | 11 ++-- .../vcs_actions/src/connection/action_service.rs | 66 ++++++++++++++++++---- 2 files changed, 61 insertions(+), 16 deletions(-) (limited to 'crates/vcs_actions') diff --git a/crates/vcs_actions/src/actions/local_actions.rs b/crates/vcs_actions/src/actions/local_actions.rs index c19b8f0..c2d9f4f 100644 --- a/crates/vcs_actions/src/actions/local_actions.rs +++ b/crates/vcs_actions/src/actions/local_actions.rs @@ -14,11 +14,10 @@ pub async fn hello_world_action(ctx: ActionContext, _n: ()) -> Result<(), TcpTar }; if ctx.is_local() { - // Invoke on local - // Send the message to the server - let _ = instance.lock().await.write_text("Hello World!").await; + // Local execution - communication is handled by on_proc_begin + info!("Hello World action executed locally"); } else if ctx.is_remote() { - // Read the message from the client + // Remote execution - read the message from the client let read = instance.lock().await.read_text().await?; info!("{}", read) } @@ -43,9 +42,9 @@ pub async fn set_upstream_vault_action( // Send the message to the server let _ = instance.lock().await.write_text("Hello World!").await; } else if ctx.is_remote() { - // Read the message from the client + // Remote execution - read the message from the client let read = instance.lock().await.read_text().await?; - info!("{}", read) + info!("Received: {}", read) } Ok(()) diff --git a/crates/vcs_actions/src/connection/action_service.rs b/crates/vcs_actions/src/connection/action_service.rs index c302fd4..f76921e 100644 --- a/crates/vcs_actions/src/connection/action_service.rs +++ b/crates/vcs_actions/src/connection/action_service.rs @@ -1,7 +1,13 @@ -use std::{net::SocketAddr, path::PathBuf, sync::Arc}; +use std::{ + net::SocketAddr, + path::PathBuf, + sync::Arc, + time::{Duration, Instant}, +}; use action_system::{action::ActionContext, action_pool::ActionPool}; use cfg_file::config::ConfigFile; +use log::{error, info, warn}; use tcp_connection::{error::TcpTargetError, instance::ConnectionInstance}; use tokio::{ net::{TcpListener, TcpStream}, @@ -67,11 +73,32 @@ fn build_server_future( let mut active_connections = 0; let mut shutdown_requested = false; - // Spawn task to handle Ctrl+C + // Spawn task to handle Ctrl+C with rapid exit detection let shutdown_tx_clone = shutdown_tx.clone(); spawn(async move { - if let Ok(()) = signal::ctrl_c().await { + let mut ctrl_c_count = 0; + let mut last_ctrl_c_time = Instant::now(); + + while let Ok(()) = signal::ctrl_c().await { + let now = Instant::now(); + + // Reset counter if more than 5 seconds have passed + if now.duration_since(last_ctrl_c_time) > Duration::from_secs(5) { + ctrl_c_count = 0; + } + + ctrl_c_count += 1; + last_ctrl_c_time = now; + let _ = shutdown_tx_clone.send(()).await; + + // If 3 Ctrl+C within 5 seconds, exit immediately + if ctrl_c_count >= 3 { + info!("Shutdown. (3/3)"); + std::process::exit(0); + } else { + info!("Ctrl + C to force shutdown. ({} / 3)", ctrl_c_count); + } } }); @@ -82,14 +109,16 @@ fn build_server_future( accept_result = listener.accept(), if !shutdown_requested => { match accept_result { Ok((stream, _addr)) => { - active_connections += 1; + info!("New connection accepted."); let _ = tx.send(1).await; let vault_clone = vault.clone(); let action_pool_clone = action_pool.clone(); let tx_clone = tx.clone(); + spawn(async move { process_connection(stream, vault_clone, action_pool_clone).await; + info!("A connection closed."); let _ = tx_clone.send(-1).await; }); } @@ -114,7 +143,10 @@ fn build_server_future( shutdown_requested = true; // If no active connections, break immediately if active_connections == 0 { + info!("No active connections. Shutting down."); break; + } else { + warn!("Cannot shutdown while active connections exist! ({} active)", active_connections); } } } @@ -131,8 +163,12 @@ async fn process_connection(stream: TcpStream, vault: Arc, action_pool: A let mut instance = ConnectionInstance::from(stream); // Read action name and action arguments - let Ok(msg) = instance.read_msgpack::().await else { - return; + let msg = match instance.read_msgpack::().await { + Ok(msg) => msg, + Err(e) => { + error!("Failed to read action message: {}", e); + return; + } }; // Build context @@ -141,11 +177,21 @@ async fn process_connection(stream: TcpStream, vault: Arc, action_pool: A // Insert vault into context let ctx = ctx.insert_arc(vault); + info!( + "Process action `{}` with argument `{}`", + msg.action_name, msg.action_args_json + ); + // Process action - let Ok(_result_json) = action_pool + let result = action_pool .process_json(&msg.action_name, ctx, msg.action_args_json) - .await - else { - return; + .await; + + match result { + Ok(_result_json) => {} + Err(e) => { + warn!("Failed to process action `{}`: {}", msg.action_name, e); + return; + } }; } -- cgit From 45cee7d8def7738d3347b201d3c54c0055817a4c Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Mon, 27 Oct 2025 17:20:01 +0800 Subject: fix: Make the ActionContext passed to on_proc_begin mutable --- crates/vcs_actions/src/registry/client_registry.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'crates/vcs_actions') diff --git a/crates/vcs_actions/src/registry/client_registry.rs b/crates/vcs_actions/src/registry/client_registry.rs index 484c4f4..a108910 100644 --- a/crates/vcs_actions/src/registry/client_registry.rs +++ b/crates/vcs_actions/src/registry/client_registry.rs @@ -26,7 +26,7 @@ pub fn client_action_pool() -> ActionPool { } async fn on_proc_begin( - ctx: &ActionContext, + ctx: &mut ActionContext, _args: &(dyn std::any::Any + Send + Sync), ) -> Result<(), TcpTargetError> { // Is ctx remote -- cgit From 2954fc5ce35126e4cf9ebfc64043c0e52a990e01 Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Mon, 27 Oct 2025 17:38:28 +0800 Subject: update: ActionContext 1. Rename `local` to `proc_on_local` 2. Add `is_remote_action` --- crates/vcs_actions/src/actions/local_actions.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) (limited to 'crates/vcs_actions') diff --git a/crates/vcs_actions/src/actions/local_actions.rs b/crates/vcs_actions/src/actions/local_actions.rs index c2d9f4f..e2788ff 100644 --- a/crates/vcs_actions/src/actions/local_actions.rs +++ b/crates/vcs_actions/src/actions/local_actions.rs @@ -13,10 +13,10 @@ pub async fn hello_world_action(ctx: ActionContext, _n: ()) -> Result<(), TcpTar )); }; - if ctx.is_local() { + if ctx.is_proc_on_local() { // Local execution - communication is handled by on_proc_begin info!("Hello World action executed locally"); - } else if ctx.is_remote() { + } else if ctx.is_proc_on_remote() { // Remote execution - read the message from the client let read = instance.lock().await.read_text().await?; info!("{}", read) @@ -37,11 +37,11 @@ pub async fn set_upstream_vault_action( )); }; - if ctx.is_local() { + if ctx.is_proc_on_local() { // Invoke on local // Send the message to the server let _ = instance.lock().await.write_text("Hello World!").await; - } else if ctx.is_remote() { + } else if ctx.is_proc_on_remote() { // Remote execution - read the message from the client let read = instance.lock().await.read_text().await?; info!("Received: {}", read) -- cgit From 3aafc7f2769284c3afab4157863e93e44c571040 Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Mon, 27 Oct 2025 17:45:01 +0800 Subject: fixed: Incorrect condition setting for determining whether to send parameters during `on_proc_begin` --- crates/vcs_actions/src/registry/client_registry.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'crates/vcs_actions') diff --git a/crates/vcs_actions/src/registry/client_registry.rs b/crates/vcs_actions/src/registry/client_registry.rs index a108910..9769750 100644 --- a/crates/vcs_actions/src/registry/client_registry.rs +++ b/crates/vcs_actions/src/registry/client_registry.rs @@ -30,7 +30,7 @@ async fn on_proc_begin( _args: &(dyn std::any::Any + Send + Sync), ) -> Result<(), TcpTargetError> { // Is ctx remote - let is_remote = ctx.is_remote(); + let is_remote = ctx.is_remote_action(); // Action name and arguments let action_name = ctx.action_name().to_string(); -- cgit From 5212da62bb6d7b3e0eb8d9b2801cc8d1d7ef4ece Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Mon, 27 Oct 2025 17:50:05 +0800 Subject: update: Remove `hello_world_action` --- crates/vcs_actions/src/actions/local_actions.rs | 21 --------------------- 1 file changed, 21 deletions(-) (limited to 'crates/vcs_actions') diff --git a/crates/vcs_actions/src/actions/local_actions.rs b/crates/vcs_actions/src/actions/local_actions.rs index e2788ff..f705692 100644 --- a/crates/vcs_actions/src/actions/local_actions.rs +++ b/crates/vcs_actions/src/actions/local_actions.rs @@ -4,27 +4,6 @@ use action_system::{action::ActionContext, macros::action_gen}; use log::info; use tcp_connection::error::TcpTargetError; -#[action_gen] -pub async fn hello_world_action(ctx: ActionContext, _n: ()) -> Result<(), TcpTargetError> { - // Ensure the instance is available - let Some(instance) = ctx.instance() else { - return Err(TcpTargetError::NotFound( - "Connection Instance Lost.".to_string(), - )); - }; - - if ctx.is_proc_on_local() { - // Local execution - communication is handled by on_proc_begin - info!("Hello World action executed locally"); - } else if ctx.is_proc_on_remote() { - // Remote execution - read the message from the client - let read = instance.lock().await.read_text().await?; - info!("{}", read) - } - - Ok(()) -} - #[action_gen] pub async fn set_upstream_vault_action( ctx: ActionContext, -- cgit From 60219b20754dda7f560deb5e9e442d46e4636507 Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Mon, 27 Oct 2025 17:50:35 +0800 Subject: update: Add server lock behavior to vault lifecycle Lock the vault before starting the server and unlock it during shutdown. --- crates/vcs_actions/src/connection/action_service.rs | 9 +++++++++ 1 file changed, 9 insertions(+) (limited to 'crates/vcs_actions') diff --git a/crates/vcs_actions/src/connection/action_service.rs b/crates/vcs_actions/src/connection/action_service.rs index f76921e..a657408 100644 --- a/crates/vcs_actions/src/connection/action_service.rs +++ b/crates/vcs_actions/src/connection/action_service.rs @@ -31,6 +31,12 @@ pub async fn server_entry(vault_path: impl Into) -> Result<(), TcpTarge // Initialize the vault let vault: Arc = init_vault(vault_cfg, vault_path.into()).await?; + // Lock the vault + vault.lock().map_err(|e| { + error!("{}", e); + TcpTargetError::Locked(e.to_string()) + })?; + // Create ActionPool let action_pool: Arc = Arc::new(server_action_pool()); @@ -38,6 +44,9 @@ pub async fn server_entry(vault_path: impl Into) -> Result<(), TcpTarge let (_shutdown_rx, future) = build_server_future(vault.clone(), action_pool.clone(), listener); future.await?; // Start and block until shutdown + // Unlock the vault + vault.unlock()?; + Ok(()) } -- cgit From b772e12c22b87e1473d1002ef9b82d954d2239a7 Mon Sep 17 00:00:00 2001 From: 魏曹先生 <1992414357@qq.com> Date: Mon, 27 Oct 2025 17:51:07 +0800 Subject: update: Add connection logs to show active count Include current connection count in connection open/close log messages for better monitoring of server load --- crates/vcs_actions/src/connection/action_service.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'crates/vcs_actions') diff --git a/crates/vcs_actions/src/connection/action_service.rs b/crates/vcs_actions/src/connection/action_service.rs index a657408..f787cae 100644 --- a/crates/vcs_actions/src/connection/action_service.rs +++ b/crates/vcs_actions/src/connection/action_service.rs @@ -118,7 +118,7 @@ fn build_server_future( accept_result = listener.accept(), if !shutdown_requested => { match accept_result { Ok((stream, _addr)) => { - info!("New connection accepted."); + info!("New connection. (now {})", active_connections); let _ = tx.send(1).await; let vault_clone = vault.clone(); @@ -127,7 +127,7 @@ fn build_server_future( spawn(async move { process_connection(stream, vault_clone, action_pool_clone).await; - info!("A connection closed."); + info!("A connection closed. (now {})", active_connections); let _ = tx_clone.send(-1).await; }); } -- cgit