Add tests, refactor error handling, and implement rate limiting

Cargo.toml

@@ -1,40 +1,40 @@
-[workspace]
-members = [
-    "cuscuta-common",
-    "cuscuta-entry",
-    "cuscuta-worker",
-    "cuscuta-chilo",
-    "cuscuta-mock",
-    "cuscutactl",
-]
-resolver = "2"
-
-[workspace.package]
-version = "0.1.8"
-
-[workspace.dependencies]
-anyhow = "1.0.102"
-axum = "0.8.9"
-base64 = "0.22.1"
-chrono = { version = "0.4.44", features = ["serde"] }
-comfy-table = "7.1"
-hex = "0.4.3"
-rand = "0.10.1"
-redis = { version = "1.2.0", features = ["streams"] }
-reqwest = { version = "0.13.2", features = ["json", "query", "form"] }
-serde = { version = "1.0.228", features = ["derive"] }
-serde_json = "1.0.150"
-sha2 = "0.11.0"
-sqlx = { version = "0.8.6", features = ["postgres", "chrono", "runtime-tokio"] }
-thiserror = "2.0.18"
-tokio = { version = "1.52.3", features = ["rt-multi-thread"] }
-tokio-util = "0.7.18"
-tracing = "0.1.44"
-tracing-subscriber = { version = "0.3.23", features = ["env-filter"] }
-
-[workspace.lints.rust]
-missing_docs = "deny"
-
-[workspace.lints.clippy]
-pedantic = "deny"
-nursery = "deny"
+[workspace]
+members = [
+    "cuscuta-common",
+    "cuscuta-entry",
+    "cuscuta-worker",
+    "cuscuta-chilo",
+    "cuscuta-mock",
+    "cuscutactl", "cuscuta-test",
+]
+resolver = "2"
+
+[workspace.package]
+version = "0.1.9"
+
+[workspace.dependencies]
+anyhow = "1.0.102"
+axum = "0.8.9"
+base64 = "0.22.1"
+chrono = { version = "0.4.44", features = ["serde"] }
+comfy-table = "7.1"
+hex = "0.4.3"
+rand = "0.10.1"
+redis = { version = "1.2.0", features = ["streams"] }
+reqwest = { version = "0.13.2", features = ["json", "query", "form"] }
+serde = { version = "1.0.228", features = ["derive"] }
+serde_json = "1.0.150"
+sha2 = "0.11.0"
+sqlx = { version = "0.8.6", features = ["postgres", "chrono", "runtime-tokio"] }
+thiserror = "2.0.18"
+tokio = { version = "1.52.3", features = ["rt-multi-thread"] }
+tokio-util = "0.7.18"
+tracing = "0.1.44"
+tracing-subscriber = { version = "0.3.23", features = ["env-filter"] }
+
+[workspace.lints.rust]
+missing_docs = "deny"
+
+[workspace.lints.clippy]
+pedantic = "deny"
+nursery = "deny"

Deployment.en.md

@@ -174,6 +174,7 @@ All cuscuta components receive configuration through environment variables.
 | `WORKER_ACCOUNT_LEASE_TIME_SECS` | 120 | Account lease duration (s) |
 | `WORKER_ACCOUNT_LEASE_TIME_REFRESH_GAP_SECS` | 30 | Lease refresh interval (s) |
 | `WORKER_JOB_MAX_WORK_TIME_SECS` | 1200 | Maximum working time for a job (s) |
+| `WORKER_EMPTY_FRIENDS_DELAY_TIME_SECS` | 10 | Rate limiting delay time (s) |
 | `ETA_ENABLE` | true | Enable ETA estimation |
 | `API_CHILO` | — | chilo service URL |
 | `API_LOGIN` | — | Login API endpoint |

Deployment.md

@@ -169,6 +169,7 @@ cuscuta 所有组件均通过环境变量注入配置。
 | `WORKER_ACCOUNT_LEASE_TIME_SECS` | 120 | 账号租约时长（秒） |
 | `WORKER_ACCOUNT_LEASE_TIME_REFRESH_GAP_SECS` | 30 | 租约续期间隔（秒） |
 | `WORKER_JOB_MAX_WORK_TIME_SECS` | 1200 | 任务的最长运行时间（秒） |
+| `WORKER_EMPTY_FRIENDS_DELAY_TIME_SECS` | 10 | 风控应对延迟（秒） |
 | `ETA_ENABLE` | true | 是否启用 ETA 预估 |
 | `API_CHILO` | — | chilo 服务地址 |
 | `API_LOGIN` | — | 登录 API 地址 |

TODO.md

@@ -1,5 +1,11 @@
 ## 计划表
 
+### 杂项
+- [x] tracing重构
+- [ ] 日志重构
+  - [ ] 日志细化
+  - [ ] 日志JSON开关
+
 ### cuscuta-worker
 
 - [x] 基础查分服务

cuscuta-common/Cargo.toml

@@ -19,5 +19,8 @@ thiserror = { workspace = true }
 tokio = { workspace = true, features = ["signal", "macros"] }
 tokio-util = { workspace = true }
 
+[dev-dependencies]
+cuscuta-test = { path = "../cuscuta-test" }
+
 [lints]
 workspace = true

cuscuta-common/src/api/chilo.rs

@@ -40,5 +40,7 @@ pub async fn chilo_generate(timestamp: &str, path: &str, kind: &str) -> Result<C
         .map_err(Error::Network)?
         .json::<ChiloResult>()
         .await
-        .map_err(|e| Error::Decode(e.to_string()))
+        .map_err(|e| Error::Decode {
+            message: e.to_string(),
+        })
 }

cuscuta-common/src/api/github.rs

@@ -1,70 +1,90 @@
-use base64::Engine;
-use reqwest::StatusCode;
-use serde::{Deserialize, de::DeserializeOwned};
-
-use crate::api::{Error, ErrorForStatusWithResponse};
-
-#[derive(Debug, Deserialize)]
-struct GitHubFileInternal {
-    content: String,
-    download_url: String,
-}
-
-type Result<T> = core::result::Result<T, Error>;
-
-/// 调用Github Api，从Github上拉取资源
-///
-/// # Errors
-/// - 当环境变量配置无效时，返回[`Error::Env`]
-/// - 当`reqwest`客户端初始化失败时，返回[`Error::ClientSetup`]
-/// - 当请求发送失败时，返回[`Error::Network`]
-/// - 当返回值不为2xx时，返回[`Error::BadStatus`]
-/// - 当Json反序列化失败时，返回[`Error::Decode`]
-pub async fn fetch_github_resource<T>(repo: &str, path: &str, token: &str) -> Result<T>
-where
-    T: DeserializeOwned,
-{
-    let file_object = reqwest::Client::builder()
-        .user_agent("curl/7.88.1")
-        .build()
-        .map_err(Error::ClientSetup)?
-        .get(format!(
-            "https://api.github.com/repos/{repo}/contents/{path}"
-        ))
-        .header("Accept", "application/vnd.github.object")
-        .header("Authorization", format!("Bearer {token}"))
-        .header("X-GitHub-Api-Version", "2026-03-10")
-        .send()
-        .await
-        .map_err(Error::Network)?
-        .error_for_status_with_response()
-        .await
-        .map_err(|(s, e)| Error::BadStatus(e.status().unwrap_or_else(StatusCode::default), s))?
-        .json::<GitHubFileInternal>()
-        .await
-        .map_err(|e| Error::Decode(format!("phase1: {e}")))?;
-    if file_object.content.is_empty() {
-        reqwest::Client::builder()
-            .user_agent("curl/7.88.1")
-            .build()
-            .map_err(Error::ClientSetup)?
-            .get(file_object.download_url)
-            .send()
-            .await
-            .map_err(Error::Network)?
-            .error_for_status_with_response()
-            .await
-            .map_err(|(s, e)| Error::BadStatus(e.status().unwrap_or_else(StatusCode::default), s))?
-            .json::<T>()
-            .await
-            .map_err(|e| Error::Decode(format!("phase b_1: {e}")))
-    } else {
-        base64::prelude::BASE64_STANDARD
-            .decode(file_object.content.replace('\n', ""))
-            .map_err(|e| Error::Decode(format!("phase a_2: {e}")))
-            .map(|it| String::from_utf8(it).map_err(|e| Error::Decode(format!("phase a_3: {e}"))))?
-            .map(|it| {
-                serde_json::from_str::<T>(&it).map_err(|e| Error::Decode(format!("phase a_4: {e}")))
-            })?
-    }
-}
+use base64::Engine;
+use reqwest::StatusCode;
+use serde::{Deserialize, de::DeserializeOwned};
+
+use crate::api::{Error, ErrorForStatusWithResponse};
+
+#[derive(Debug, Deserialize)]
+struct GitHubFileInternal {
+    content: String,
+    download_url: String,
+}
+
+type Result<T> = core::result::Result<T, Error>;
+
+/// 调用Github Api，从Github上拉取资源
+///
+/// # Errors
+/// - 当环境变量配置无效时，返回[`Error::Env`]
+/// - 当`reqwest`客户端初始化失败时，返回[`Error::ClientSetup`]
+/// - 当请求发送失败时，返回[`Error::Network`]
+/// - 当返回值不为2xx时，返回[`Error::BadStatus`]
+/// - 当Json反序列化失败时，返回[`Error::Decode`]
+pub async fn fetch_github_resource<T>(repo: &str, path: &str, token: &str) -> Result<T>
+where
+    T: DeserializeOwned,
+{
+    let file_object = reqwest::Client::builder()
+        .user_agent("curl/7.88.1")
+        .build()
+        .map_err(Error::ClientSetup)?
+        .get(format!(
+            "https://api.github.com/repos/{repo}/contents/{path}"
+        ))
+        .header("Accept", "application/vnd.github.object")
+        .header("Authorization", format!("Bearer {token}"))
+        .header("X-GitHub-Api-Version", "2026-03-10")
+        .send()
+        .await
+        .map_err(Error::Network)?
+        .error_for_status_with_response()
+        .await
+        .map_err(|(s, e)| Error::BadStatus {
+            status_code: e.status().unwrap_or_else(StatusCode::default),
+            message: s,
+            extra_error_code: None,
+        })?
+        .json::<GitHubFileInternal>()
+        .await
+        .map_err(|e| Error::Decode {
+            message: format!("phase1: {e}"),
+        })?;
+    if file_object.content.is_empty() {
+        reqwest::Client::builder()
+            .user_agent("curl/7.88.1")
+            .build()
+            .map_err(Error::ClientSetup)?
+            .get(file_object.download_url)
+            .send()
+            .await
+            .map_err(Error::Network)?
+            .error_for_status_with_response()
+            .await
+            .map_err(|(s, e)| Error::BadStatus {
+                status_code: e.status().unwrap_or_else(StatusCode::default),
+                message: s,
+                extra_error_code: None,
+            })?
+            .json::<T>()
+            .await
+            .map_err(|e| Error::Decode {
+                message: format!("phase b_1: {e}"),
+            })
+    } else {
+        base64::prelude::BASE64_STANDARD
+            .decode(file_object.content.replace('\n', ""))
+            .map_err(|e| Error::Decode {
+                message: format!("phase a_2: {e}"),
+            })
+            .map(|it| {
+                String::from_utf8(it).map_err(|e| Error::Decode {
+                    message: format!("phase a_3: {e}"),
+                })
+            })?
+            .map(|it| {
+                serde_json::from_str::<T>(&it).map_err(|e| Error::Decode {
+                    message: format!("phase a_4: {e}"),
+                })
+            })?
+    }
+}

cuscuta-common/src/api/mod.rs

@@ -12,6 +12,9 @@ pub mod xxxxxx;
 pub mod chilo;
 
 /// Api调用可能引发的错误
+///
+/// 注：这里的错误处理可能很脏，因为这个错误类型包含了过于特化的[`Self::TooManyRetries`]
+/// 以及并非所有Api系函数均支持的[`Self::ApiError`]
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
     /// `reqwest`客户端初始化失败
@@ -23,24 +26,41 @@ pub enum Error {
     Network(reqwest::Error),
 
     /// Api的`HTTP`返回码不为2xx或1xx
-    #[error("bad return status: {0}: {1}")]
-    BadStatus(StatusCode, String),
+    #[error("bad return status: HTTP {status_code} {extra_error_code:?}: {message}")]
+    BadStatus {
+        /// 错误码
+        status_code: StatusCode,
+
+        /// 错误描述
+        message: String,
+
+        /// Api错误码（如果有）
+        extra_error_code: Option<i64>,
+    },
 
     /// Json反序列化失败
-    #[error("failed to decode response: {0}")]
-    Decode(String),
+    #[error("failed to decode response: {message}")]
+    Decode {
+        /// 错误描述
+        message: String,
+    },
 
     /// 环境变量未配置或配置无效
-    #[error("failed to read env::var: {0}:{1}")]
-    Env(env::VarError, String),
+    #[error("failed to read env::var: {error}:{message}")]
+    Env {
+        /// 环境变量错误
+        error: env::VarError,
 
-    /// 重试次数过多
-    #[error("too many retries: inner: {0}")]
-    TooManyRetries(String),
+        /// 错误描述
+        message: String,
+    },
 }
 
 fn try_get_env_var(var: &str) -> Result<String, Error> {
-    env::var(var).map_err(|e| Error::Env(e, var.to_string()))
+    env::var(var).map_err(|error| Error::Env {
+        error,
+        message: var.to_string(),
+    })
 }
 
 trait ErrorForStatusWithResponse