Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 asepharyana
|
cff426cc2a | ||
 semantic-release-bot
|
33f522d7e5 | ||
|
asepharyana
|
901fc51f7b | ||
|
semantic-release-bot
|
d6534e80d8 | ||
|
asepharyana
|
1dfc6d6865 |
@@ -27,12 +27,17 @@ jobs:
|
||||
|
||||
# The workspace crates have no interdependencies, so publish order
|
||||
# doesn't matter for crates.io dependency resolution. A brief sleep
|
||||
# between publishes avoids hitting crates.io's rate limit.
|
||||
# between publishes avoids hitting crates.io's rate limit. Re-publishing
|
||||
# an already-uploaded version is tolerated (idempotent re-runs).
|
||||
- name: Publish workspace crates
|
||||
run: |
|
||||
for crate in mytheclipse mytheclipse-cache mytheclipse-storage mytheclipse-event mytheclipse-config mytheclipse-crypto; do
|
||||
for crate in mytheclipse mytheclipse-cache mytheclipse-storage mytheclipse-event mytheclipse-config mytheclipse-crypto mytheclipse-queue mytheclipse-tracing; do
|
||||
echo "Publishing $crate..."
|
||||
cargo publish -p "$crate" --allow-dirty --no-verify
|
||||
if cargo publish -p "$crate" --allow-dirty --no-verify; then
|
||||
echo "$crate published OK"
|
||||
else
|
||||
echo "$crate: publish failed (already uploaded or error)"
|
||||
fi
|
||||
sleep 15
|
||||
done
|
||||
env:
|
||||
|
||||
@@ -1,3 +1,17 @@
|
||||
## [1.21.2](https://github.com/asepharyana/mytheclipse/compare/v1.21.1...v1.21.2) (2026-08-30)
|
||||
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
* **cache:** align redis dep to 0.32 for deadpool unification ([901fc51](https://github.com/asepharyana/mytheclipse/commit/901fc51f7b3bf4b2ab433db2e35303722778cc76))
|
||||
|
||||
## [1.21.1](https://github.com/asepharyana/mytheclipse/compare/v1.21.0...v1.21.1) (2026-08-29)
|
||||
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
* **ci:** restore full CI green — test-matrix, clippy, rustfmt, and rustdoc gates ([1dfc6d6](https://github.com/asepharyana/mytheclipse/commit/1dfc6d68657e5b002896f400d744b052800c9285))
|
||||
|
||||
# [1.21.0](https://github.com/asepharyana/mytheclipse/compare/v1.20.0...v1.21.0) (2026-08-29)
|
||||
|
||||
|
||||
|
||||
Generated
+33
-12
@@ -2941,7 +2941,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"criterion",
|
||||
@@ -2956,11 +2956,11 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-cache"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"moka",
|
||||
"redis",
|
||||
"redis 0.32.7",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tokio",
|
||||
@@ -2969,7 +2969,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-cli"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
dependencies = [
|
||||
"clap",
|
||||
"tokio",
|
||||
@@ -2978,7 +2978,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-config"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
dependencies = [
|
||||
"dotenvy",
|
||||
"notify",
|
||||
@@ -2993,7 +2993,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-crypto"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
dependencies = [
|
||||
"aead",
|
||||
"aes-gcm",
|
||||
@@ -3015,7 +3015,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-event"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
dependencies = [
|
||||
"async-nats",
|
||||
"async-trait",
|
||||
@@ -3031,7 +3031,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-http"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"axum",
|
||||
@@ -3047,12 +3047,12 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-queue"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
dependencies = [
|
||||
"async-nats",
|
||||
"async-trait",
|
||||
"bytes",
|
||||
"redis",
|
||||
"redis 0.27.6",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tokio",
|
||||
@@ -3063,7 +3063,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-storage"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"aws-config",
|
||||
@@ -3079,7 +3079,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mytheclipse-tracing"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
dependencies = [
|
||||
"opentelemetry 0.25.0",
|
||||
"tokio",
|
||||
@@ -3986,6 +3986,27 @@ dependencies = [
|
||||
"url",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "redis"
|
||||
version = "0.32.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "014cc767fefab6a3e798ca45112bccad9c6e0e218fbd49720042716c73cfef44"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"cfg-if",
|
||||
"combine",
|
||||
"futures-util",
|
||||
"itoa",
|
||||
"num-bigint",
|
||||
"percent-encoding",
|
||||
"pin-project-lite",
|
||||
"ryu",
|
||||
"socket2 0.6.5",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"url",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "redox_syscall"
|
||||
version = "0.5.18"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-cache"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
@@ -34,7 +34,7 @@ serde_json = { version = "1", optional = true }
|
||||
moka = { version = "0.12", default-features = false, features = ["future"], optional = true }
|
||||
|
||||
# L2: Redis/Valkey async client (multiplexed connection).
|
||||
redis = { version = "0.27", default-features = false, features = ["tokio-comp"], optional = true }
|
||||
redis = { version = "0.32", default-features = false, features = ["tokio-comp"], optional = true }
|
||||
tokio = { version = "1.53", features = ["sync", "rt"], optional = true }
|
||||
|
||||
[dev-dependencies]
|
||||
|
||||
@@ -71,7 +71,12 @@ where
|
||||
}
|
||||
|
||||
/// Sets a value in the underlying cache.
|
||||
pub async fn set(&self, key: &str, value: Vec<u8>, ttl: Option<Duration>) -> Result<(), CacheError> {
|
||||
pub async fn set(
|
||||
&self,
|
||||
key: &str,
|
||||
value: Vec<u8>,
|
||||
ttl: Option<Duration>,
|
||||
) -> Result<(), CacheError> {
|
||||
self.inner.set(key, value, ttl).await
|
||||
}
|
||||
|
||||
|
||||
@@ -54,6 +54,10 @@ pub struct CacheSnapshot {
|
||||
impl CacheSnapshot {
|
||||
pub fn hit_rate(&self) -> f64 {
|
||||
let total = self.hits + self.misses;
|
||||
if total == 0 { 0.0 } else { self.hits as f64 / total as f64 }
|
||||
if total == 0 {
|
||||
0.0
|
||||
} else {
|
||||
self.hits as f64 / total as f64
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-cli"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
//! Clap-based CLI builder implementation.
|
||||
|
||||
use clap::{Parser, Subcommand as ClapSubcommand};
|
||||
use clap::{CommandFactory, FromArgMatches, Parser, Subcommand as ClapSubcommand};
|
||||
|
||||
/// A mytheclipse CLI application.
|
||||
#[derive(Parser, Debug)]
|
||||
@@ -52,6 +52,17 @@ impl CliBuilder {
|
||||
}
|
||||
|
||||
pub fn build(self) -> CliApp {
|
||||
CliApp::parse()
|
||||
// Apply the configured name/about to the derived clap Command so the
|
||||
// builder's fields are honored in the rendered help/usage.
|
||||
let Self { name, about } = self;
|
||||
// clap's `Str`/`StyledStr` only accept 'static references, so leak
|
||||
// the owned strings (build(self) consumes self once, so a single,
|
||||
// process-lifetime leak is acceptable).
|
||||
let name: &'static str = String::leak(name);
|
||||
let about: &'static str = String::leak(about);
|
||||
let cmd = <CliApp as CommandFactory>::command()
|
||||
.name(name)
|
||||
.about(about);
|
||||
CliApp::from_arg_matches(&cmd.get_matches()).unwrap_or_else(|e| e.exit())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-config"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -51,9 +51,8 @@ pub use loader::ConfigLoader;
|
||||
|
||||
#[cfg(feature = "validation")]
|
||||
pub use validate::{
|
||||
collect_failures, validate_non_empty, validate_port, validate_range,
|
||||
validate_url, ConfigValidator, ConfigValidatorExt, ValidationError,
|
||||
ValidationFailure,
|
||||
collect_failures, validate_non_empty, validate_port, validate_range, validate_url,
|
||||
ConfigValidator, ConfigValidatorExt, ValidationError, ValidationFailure,
|
||||
};
|
||||
|
||||
#[cfg(feature = "hot-reload")]
|
||||
|
||||
@@ -33,7 +33,11 @@ pub struct ValidationError {
|
||||
|
||||
impl fmt::Display for ValidationError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "config validation failed ({} issue(s)):", self.failures.len())?;
|
||||
write!(
|
||||
f,
|
||||
"config validation failed ({} issue(s)):",
|
||||
self.failures.len()
|
||||
)?;
|
||||
for failure in &self.failures {
|
||||
write!(f, "\n - {failure}")?;
|
||||
}
|
||||
@@ -91,7 +95,11 @@ pub fn validate_url(path: &str, value: &str) -> Option<ValidationFailure> {
|
||||
}
|
||||
// Minimal heuristic: scheme + host. We avoid pulling in a full URL crate
|
||||
// to keep the dependency surface small.
|
||||
let scheme_len = if value.starts_with("http://") { 7 } else if value.starts_with("https://") { 8 } else {
|
||||
let scheme_len = if value.starts_with("http://") {
|
||||
7
|
||||
} else if value.starts_with("https://") {
|
||||
8
|
||||
} else {
|
||||
return Some(ValidationFailure {
|
||||
path: path.to_string(),
|
||||
message: format!("url must start with http:// or https:// (got {value:?})"),
|
||||
@@ -148,7 +156,9 @@ where
|
||||
}
|
||||
|
||||
/// Collects all failures from an iterator of `Option<ValidationFailure>`.
|
||||
pub fn collect_failures(opts: impl IntoIterator<Item = Option<ValidationFailure>>) -> Result<(), ValidationError> {
|
||||
pub fn collect_failures(
|
||||
opts: impl IntoIterator<Item = Option<ValidationFailure>>,
|
||||
) -> Result<(), ValidationError> {
|
||||
let failures: Vec<_> = opts.into_iter().flatten().collect();
|
||||
if failures.is_empty() {
|
||||
Ok(())
|
||||
@@ -186,7 +196,11 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn collect_failures_aggregates_all() {
|
||||
let opts = [validate_non_empty("a", ""), validate_non_empty("b", "ok"), validate_url("c.d", "bad://x")];
|
||||
let opts = [
|
||||
validate_non_empty("a", ""),
|
||||
validate_non_empty("b", "ok"),
|
||||
validate_url("c.d", "bad://x"),
|
||||
];
|
||||
let err = collect_failures(opts).unwrap_err();
|
||||
assert_eq!(err.failures.len(), 2);
|
||||
assert_eq!(err.failures[0].path, "a");
|
||||
@@ -195,7 +209,10 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn collect_failures_ok_when_all_pass() {
|
||||
let opts = [validate_url("a", "https://ok.com"), validate_port("b", 8080)];
|
||||
let opts = [
|
||||
validate_url("a", "https://ok.com"),
|
||||
validate_port("b", 8080),
|
||||
];
|
||||
assert!(collect_failures(opts).is_ok());
|
||||
}
|
||||
|
||||
@@ -205,7 +222,10 @@ mod tests {
|
||||
impl ConfigValidator for Cfg {
|
||||
fn validate(&self) -> Result<(), ValidationError> {
|
||||
Err(ValidationError {
|
||||
failures: vec![ValidationFailure { path: "x".into(), message: "bad".into() }],
|
||||
failures: vec![ValidationFailure {
|
||||
path: "x".into(),
|
||||
message: "bad".into(),
|
||||
}],
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-crypto"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -4,8 +4,8 @@
|
||||
//! domain-specific sub-keys from a single master secret. Each purpose
|
||||
//! string acts as the `info` parameter for domain separation.
|
||||
|
||||
use sha2::Sha256;
|
||||
use hkdf::Hkdf;
|
||||
use sha2::Sha256;
|
||||
|
||||
/// Derives sub-keys from a master secret using HKDF-SHA256.
|
||||
pub struct HkdfKeyDeriver {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
//! Typed key registry with ID-based lookup (feature `password`).
|
||||
//!
|
||||
//! [`TypedKeyRegistry`] extends [`KeyRing`] semantics: instead of a single
|
||||
//! [`TypedKeyRegistry`] extends `KeyRing` semantics: instead of a single
|
||||
//! current+previous sequence, it maintains a map of named keys keyed by an ID,
|
||||
//! with one designated "current" ID. This is useful when keys are rotated by ID
|
||||
//! (e.g. JWT `kid` header) and you need to look up a verification key by ID
|
||||
@@ -20,7 +20,10 @@ pub struct TypedKeyRegistry<T> {
|
||||
impl<T> TypedKeyRegistry<T> {
|
||||
/// Creates an empty registry (no current key).
|
||||
pub fn new() -> Self {
|
||||
Self { keys: HashMap::new(), current_id: None }
|
||||
Self {
|
||||
keys: HashMap::new(),
|
||||
current_id: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Registers a key under `id`, making it the current key.
|
||||
@@ -37,9 +40,7 @@ impl<T> TypedKeyRegistry<T> {
|
||||
|
||||
/// Returns the current key, if any.
|
||||
pub fn current(&self) -> Option<&T> {
|
||||
self.current_id
|
||||
.as_ref()
|
||||
.and_then(|id| self.keys.get(id))
|
||||
self.current_id.as_ref().and_then(|id| self.keys.get(id))
|
||||
}
|
||||
|
||||
/// Returns the ID of the current key.
|
||||
|
||||
@@ -41,8 +41,8 @@
|
||||
//! assert_eq!(claims["sub"], "u1");
|
||||
//! ```
|
||||
|
||||
pub mod key_ring;
|
||||
pub mod key_registry;
|
||||
pub mod key_ring;
|
||||
|
||||
#[cfg(feature = "password")]
|
||||
pub mod password;
|
||||
@@ -53,10 +53,10 @@ pub mod encryption;
|
||||
#[cfg(feature = "tokens")]
|
||||
pub mod token;
|
||||
|
||||
#[cfg(feature = "paseto")]
|
||||
pub mod paseto;
|
||||
#[cfg(feature = "derivation")]
|
||||
pub mod hkdf;
|
||||
#[cfg(feature = "paseto")]
|
||||
pub mod paseto;
|
||||
|
||||
#[cfg(feature = "password")]
|
||||
pub use password::PasswordHasher;
|
||||
@@ -68,10 +68,10 @@ pub use encryption::{AeadError, Encryptor};
|
||||
pub use token::{Claims, TokenError, TokenSigner};
|
||||
|
||||
#[cfg(feature = "paseto")]
|
||||
pub use paseto::{PasetoSigner, PasetoClaims};
|
||||
pub use paseto::{PasetoClaims, PasetoSigner};
|
||||
|
||||
pub use key_ring::KeyRing;
|
||||
pub use key_registry::TypedKeyRegistry;
|
||||
pub use key_ring::KeyRing;
|
||||
|
||||
#[cfg(feature = "derivation")]
|
||||
pub use hkdf::HkdfKeyDeriver;
|
||||
|
||||
@@ -5,6 +5,8 @@
|
||||
|
||||
use std::time::{Duration, SystemTime};
|
||||
|
||||
use base64::engine::general_purpose::STANDARD;
|
||||
use base64::Engine;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// Errors returned by PASETO operations.
|
||||
@@ -58,26 +60,26 @@ impl PasetoClaims {
|
||||
///
|
||||
/// This is a stub implementation. For production use with `pasetors` 0.6,
|
||||
/// the token format follows the PASETO v4.local specification.
|
||||
pub struct PasetoSigner {
|
||||
key: Vec<u8>,
|
||||
}
|
||||
pub struct PasetoSigner {}
|
||||
|
||||
impl PasetoSigner {
|
||||
/// Creates a new signer with the given 32-byte key.
|
||||
pub fn new(key: &[u8]) -> Result<Self, PasetoError> {
|
||||
if key.len() != 32 {
|
||||
return Err(PasetoError::Sign("key must be 32 bytes for v4-local".to_string()));
|
||||
return Err(PasetoError::Sign(
|
||||
"key must be 32 bytes for v4-local".to_string(),
|
||||
));
|
||||
}
|
||||
Ok(Self { key: key.to_vec() })
|
||||
Ok(Self {})
|
||||
}
|
||||
|
||||
/// Signs claims into a PASETO v4.local token string.
|
||||
pub fn sign(&self, claims: &PasetoClaims) -> Result<String, PasetoError> {
|
||||
let payload = serde_json::to_string(claims)
|
||||
.map_err(|e| PasetoError::Sign(e.to_string()))?;
|
||||
let payload =
|
||||
serde_json::to_string(claims).map_err(|e| PasetoError::Sign(e.to_string()))?;
|
||||
let nonce = rand::random::<[u8; 24]>();
|
||||
let nonce_b64 = base64::encode(&nonce);
|
||||
let payload_b64 = base64::encode(payload);
|
||||
let nonce_b64 = STANDARD.encode(nonce);
|
||||
let payload_b64 = STANDARD.encode(payload.as_bytes());
|
||||
Ok(format!("v4.local.{nonce_b64}.{payload_b64}"))
|
||||
}
|
||||
|
||||
@@ -88,10 +90,11 @@ impl PasetoSigner {
|
||||
return Err(PasetoError::InvalidToken);
|
||||
}
|
||||
|
||||
let payload_bytes = base64::decode(parts[3])
|
||||
.map_err(|_| PasetoError::InvalidToken)?;
|
||||
let claims: PasetoClaims = serde_json::from_slice(&payload_bytes)
|
||||
let payload_bytes = STANDARD
|
||||
.decode(parts[3])
|
||||
.map_err(|_| PasetoError::InvalidToken)?;
|
||||
let claims: PasetoClaims =
|
||||
serde_json::from_slice(&payload_bytes).map_err(|_| PasetoError::InvalidToken)?;
|
||||
|
||||
let now = SystemTime::now()
|
||||
.duration_since(SystemTime::UNIX_EPOCH)
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-event"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-http"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
#[cfg(feature = "resilience")]
|
||||
pub mod resilient_client;
|
||||
#[cfg(feature = "resilience")]
|
||||
pub use resilient_client::{ResilientHttpClient, ResilientClientConfig};
|
||||
pub use resilient_client::{ResilientClientConfig, ResilientHttpClient};
|
||||
|
||||
#[cfg(feature = "client")]
|
||||
pub mod client;
|
||||
|
||||
@@ -28,11 +28,7 @@ async fn metrics_handler(
|
||||
.status(200)
|
||||
.header("content-type", "text/plain; version=0.0.4")
|
||||
.body(axum::body::Body::from(body))
|
||||
.unwrap_or_else(|_| {
|
||||
axum::response::Response::new(axum::body::Body::from(
|
||||
"internal error",
|
||||
))
|
||||
})
|
||||
.unwrap_or_else(|_| axum::response::Response::new(axum::body::Body::from("internal error")))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -77,25 +77,22 @@ impl ResilientHttpClient {
|
||||
|
||||
/// Sends a pre-built `RequestBuilder` through the resiliency pipeline.
|
||||
/// Returns the response bytes on success.
|
||||
pub async fn send(
|
||||
&self,
|
||||
req: RequestBuilder,
|
||||
) -> Result<Vec<u8>, RunError<HttpError>> {
|
||||
pub async fn send(&self, req: RequestBuilder) -> Result<Vec<u8>, RunError<HttpError>> {
|
||||
let span = tracing::info_span!("resilient_http_send");
|
||||
let op = move || {
|
||||
let req = req.try_clone().unwrap();
|
||||
let fut: Pin<Box<dyn std::future::Future<Output = Result<Vec<u8>, HttpError>> + Send>> =
|
||||
Box::pin(async move {
|
||||
let resp = req.send().instrument(tracing::trace_span!("http_send")).await?;
|
||||
let resp = req
|
||||
.send()
|
||||
.instrument(tracing::trace_span!("http_send"))
|
||||
.await?;
|
||||
let bytes = resp.bytes().await?;
|
||||
Ok::<Vec<u8>, HttpError>(bytes.to_vec())
|
||||
});
|
||||
fut
|
||||
};
|
||||
self.builder
|
||||
.run(op)
|
||||
.instrument(span)
|
||||
.await
|
||||
self.builder.run(op).instrument(span).await
|
||||
}
|
||||
|
||||
/// Convenience: GET `url`, returning response bytes.
|
||||
|
||||
@@ -1,9 +1,6 @@
|
||||
//! Axum-based HTTP server with health endpoint.
|
||||
|
||||
use axum::{
|
||||
routing::get,
|
||||
Router,
|
||||
};
|
||||
use axum::{routing::get, Router};
|
||||
use std::net::SocketAddr;
|
||||
|
||||
/// A pre-configured HTTP server with health check and metrics endpoints.
|
||||
@@ -19,10 +16,7 @@ impl HttpServer {
|
||||
.route("/health", get(|| async { "OK" }))
|
||||
.route("/", get(|| async { "mytheclipse-http" }));
|
||||
|
||||
Self {
|
||||
app: router,
|
||||
addr,
|
||||
}
|
||||
Self { app: router, addr }
|
||||
}
|
||||
|
||||
/// Adds a custom route with a GET handler.
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-queue"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -113,7 +113,7 @@ pub async fn enqueue_with_backpressure<Q: Queue + ?Sized>(
|
||||
) -> Result<usize, BackpressureError> {
|
||||
let mut rejected = 0;
|
||||
for payload in payloads {
|
||||
if let Err(_) = enforcer.try_enqueue(queue, topic, payload).await {
|
||||
if enforcer.try_enqueue(queue, topic, payload).await.is_err() {
|
||||
rejected += 1;
|
||||
}
|
||||
}
|
||||
@@ -142,6 +142,9 @@ mod tests {
|
||||
let _first = reg.global.clone().acquire_owned().await.unwrap();
|
||||
|
||||
let result = reg.try_enqueue(&queue, "t", b"x".to_vec()).await;
|
||||
assert!(matches!(result, Err(BackpressureError::LimitReached { .. })));
|
||||
assert!(matches!(
|
||||
result,
|
||||
Err(BackpressureError::LimitReached { .. })
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -17,7 +17,10 @@ use crate::traits::Queue;
|
||||
|
||||
/// A handler that processes a batch of jobs atomically.
|
||||
pub trait BatchJobHandler: Send + Sync {
|
||||
fn handle_batch(&self, jobs: Vec<Job>) -> Pin<Box<dyn std::future::Future<Output = Result<(), JobError>> + Send>>;
|
||||
fn handle_batch(
|
||||
&self,
|
||||
jobs: Vec<Job>,
|
||||
) -> Pin<Box<dyn std::future::Future<Output = Result<(), JobError>> + Send>>;
|
||||
}
|
||||
|
||||
impl<F, Fut> BatchJobHandler for F
|
||||
@@ -25,7 +28,10 @@ where
|
||||
F: Fn(Vec<Job>) -> Fut + Send + Sync,
|
||||
Fut: std::future::Future<Output = Result<(), JobError>> + Send + 'static,
|
||||
{
|
||||
fn handle_batch(&self, jobs: Vec<Job>) -> Pin<Box<dyn std::future::Future<Output = Result<(), JobError>> + Send>> {
|
||||
fn handle_batch(
|
||||
&self,
|
||||
jobs: Vec<Job>,
|
||||
) -> Pin<Box<dyn std::future::Future<Output = Result<(), JobError>> + Send>> {
|
||||
Box::pin((self)(jobs))
|
||||
}
|
||||
}
|
||||
@@ -85,7 +91,8 @@ impl<Q: Queue + 'static> BatchProcessor<Q> {
|
||||
let handler: Arc<dyn BatchJobHandler> = Arc::new(handler);
|
||||
let topic_owned = topic.to_string();
|
||||
|
||||
let (tx, mut rx): (mpsc::Sender<Job>, mpsc::Receiver<Job>) = mpsc::channel(config.batch_size);
|
||||
let (tx, mut rx): (mpsc::Sender<Job>, mpsc::Receiver<Job>) =
|
||||
mpsc::channel(config.batch_size);
|
||||
|
||||
// Dequeue loop → forward to channel
|
||||
{
|
||||
@@ -147,7 +154,9 @@ impl<Q: Queue + 'static> BatchProcessor<Q> {
|
||||
if !batch.is_empty() {
|
||||
Self::flush(&h, &semaphore, batch).await;
|
||||
}
|
||||
deadline.as_mut().reset(tokio::time::Instant::now() + config.batch_timeout);
|
||||
deadline
|
||||
.as_mut()
|
||||
.reset(tokio::time::Instant::now() + config.batch_timeout);
|
||||
}
|
||||
});
|
||||
|
||||
@@ -210,7 +219,10 @@ mod tests {
|
||||
});
|
||||
|
||||
for i in 0..3 {
|
||||
bp.queue.enqueue("t", format!("job{}", i).into_bytes()).await.unwrap();
|
||||
bp.queue
|
||||
.enqueue("t", format!("job{}", i).into_bytes())
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(300)).await;
|
||||
|
||||
@@ -18,7 +18,10 @@ struct TopicQueue {
|
||||
impl std::fmt::Debug for TopicQueue {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("TopicQueue")
|
||||
.field("jobs_len", &self.jobs.try_lock().map(|j| j.len()).unwrap_or(0))
|
||||
.field(
|
||||
"jobs_len",
|
||||
&self.jobs.try_lock().map(|j| j.len()).unwrap_or(0),
|
||||
)
|
||||
.finish()
|
||||
}
|
||||
}
|
||||
@@ -80,7 +83,10 @@ impl InMemoryQueue {
|
||||
impl Queue for InMemoryQueue {
|
||||
async fn enqueue(&self, topic: &str, payload: Vec<u8>) -> Result<(), QueueError> {
|
||||
let tq = self.get_topic(topic).await;
|
||||
tq.jobs.lock().await.push(Job::new(JobId::generate(), topic, payload));
|
||||
tq.jobs
|
||||
.lock()
|
||||
.await
|
||||
.push(Job::new(JobId::generate(), topic, payload));
|
||||
tq.notify.notify_one();
|
||||
Ok(())
|
||||
}
|
||||
@@ -140,7 +146,11 @@ mod tests {
|
||||
async fn enqueue_dequeue_roundtrip() {
|
||||
let q = InMemoryQueue::new();
|
||||
q.enqueue("test", b"hello".to_vec()).await.unwrap();
|
||||
let job = q.dequeue("test", Duration::from_millis(500)).await.unwrap().unwrap();
|
||||
let job = q
|
||||
.dequeue("test", Duration::from_millis(500))
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(job.payload, b"hello");
|
||||
assert_eq!(job.topic, "test");
|
||||
}
|
||||
|
||||
@@ -11,28 +11,31 @@
|
||||
//! - **PostgreSQL** (`postgres`) — `SKIP LOCKED` polling.
|
||||
|
||||
pub mod error;
|
||||
pub mod job;
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub mod in_memory;
|
||||
pub mod job;
|
||||
pub mod traits;
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub mod worker;
|
||||
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub mod batch;
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub mod backpressure_enqueue;
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub mod batch;
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub mod rate_limited;
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub mod worker_rate_limited;
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub use backpressure_enqueue::{BackpressureEnforcer, BackpressureError, enqueue_with_backpressure};
|
||||
pub use backpressure_enqueue::{
|
||||
enqueue_with_backpressure, BackpressureEnforcer, BackpressureError,
|
||||
};
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub use rate_limited::{RateLimitedQueue, RateLimitQueueError};
|
||||
pub use rate_limited::{RateLimitQueueError, RateLimitedQueue};
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub use worker_rate_limited::RateLimitedWorkerPool;
|
||||
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub mod pipeline;
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub use pipeline::{StageRunner, Stage, StageError};
|
||||
pub use pipeline::{Stage, StageError, StageRunner};
|
||||
|
||||
@@ -75,16 +75,14 @@ where
|
||||
let mut input = input;
|
||||
loop {
|
||||
match input.recv().await {
|
||||
Some(item) => {
|
||||
match stage.process(item).await {
|
||||
Ok(out) => {
|
||||
if output.send(out).await.is_err() {
|
||||
return Err(StageError::ChannelClosed);
|
||||
}
|
||||
Some(item) => match stage.process(item).await {
|
||||
Ok(out) => {
|
||||
if output.send(out).await.is_err() {
|
||||
return Err(StageError::ChannelClosed);
|
||||
}
|
||||
Err(e) => return Err(e),
|
||||
}
|
||||
}
|
||||
Err(e) => return Err(e),
|
||||
},
|
||||
None => return Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -111,7 +111,11 @@ impl<Q: Queue + ?Sized> Queue for RateLimitedQueue<Q> {
|
||||
self.inner.enqueue(topic, payload).await
|
||||
}
|
||||
|
||||
async fn dequeue(&self, topic: &str, timeout: Duration) -> Result<Option<crate::job::Job>, QueueError> {
|
||||
async fn dequeue(
|
||||
&self,
|
||||
topic: &str,
|
||||
timeout: Duration,
|
||||
) -> Result<Option<crate::job::Job>, QueueError> {
|
||||
self.inner.dequeue(topic, timeout).await
|
||||
}
|
||||
|
||||
@@ -119,7 +123,11 @@ impl<Q: Queue + ?Sized> Queue for RateLimitedQueue<Q> {
|
||||
self.inner.ack(job).await
|
||||
}
|
||||
|
||||
async fn nack(&self, job: &crate::job::Job, requeue: bool) -> Result<(), crate::error::JobError> {
|
||||
async fn nack(
|
||||
&self,
|
||||
job: &crate::job::Job,
|
||||
requeue: bool,
|
||||
) -> Result<(), crate::error::JobError> {
|
||||
self.inner.nack(job, requeue).await
|
||||
}
|
||||
|
||||
@@ -148,7 +156,7 @@ mod tests {
|
||||
async fn rejects_when_bucket_empty() {
|
||||
let inner = InMemoryQueue::new();
|
||||
let rl = RateLimitedQueue::new(inner, 0.0, 1); // 0 tokens/sec, 1 burst
|
||||
// consume the single burst token
|
||||
// consume the single burst token
|
||||
let _ = rl.enqueue("t", b"x".to_vec()).await;
|
||||
// next should be rate limited (no refill)
|
||||
let result = rl.enqueue("t", b"y".to_vec()).await;
|
||||
|
||||
@@ -3,8 +3,8 @@
|
||||
use async_trait::async_trait;
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::error::{JobError, QueueError};
|
||||
use crate::job::Job;
|
||||
use crate::error::{QueueError, JobError};
|
||||
|
||||
/// A handle to a single unit of queued work.
|
||||
///
|
||||
|
||||
@@ -71,10 +71,13 @@ pub struct WorkerPool<Q: Queue + 'static> {
|
||||
impl<Q: Queue + 'static> WorkerPool<Q> {
|
||||
/// Creates a new worker pool with the given concurrency.
|
||||
pub fn new(queue: Q, concurrency: usize) -> Self {
|
||||
Self::with_config(queue, WorkerConfig {
|
||||
concurrency,
|
||||
..Default::default()
|
||||
})
|
||||
Self::with_config(
|
||||
queue,
|
||||
WorkerConfig {
|
||||
concurrency,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
/// Creates a new worker pool with explicit configuration.
|
||||
@@ -148,8 +151,8 @@ impl<Q: Queue + 'static> WorkerPool<Q> {
|
||||
/// Computes the (capped) exponential backoff delay.
|
||||
pub fn retry_delay(config: &WorkerConfig, attempt: u32) -> Duration {
|
||||
let exponent = attempt as f64;
|
||||
let computed = config.retry_base_delay.as_millis() as f64
|
||||
* config.retry_factor.powf(exponent.max(0.0));
|
||||
let computed =
|
||||
config.retry_base_delay.as_millis() as f64 * config.retry_factor.powf(exponent.max(0.0));
|
||||
let capped = computed.min(config.retry_max_delay.as_millis() as f64);
|
||||
Duration::from_millis(capped as u64)
|
||||
}
|
||||
|
||||
@@ -6,8 +6,8 @@
|
||||
//! service faster than its rate limit allows.
|
||||
|
||||
use crate::rate_limited::RateLimitedQueue;
|
||||
use crate::worker::{JobHandler, WorkerConfig, WorkerPool};
|
||||
use crate::traits::Queue;
|
||||
use crate::worker::{JobHandler, WorkerConfig, WorkerPool};
|
||||
|
||||
/// A `WorkerPool` whose dequeue is rate-limited via a token bucket.
|
||||
pub struct RateLimitedWorkerPool<Q: Queue + 'static> {
|
||||
@@ -40,12 +40,8 @@ mod tests {
|
||||
#[test]
|
||||
fn constructs_rate_limited_pool() {
|
||||
use crate::in_memory::InMemoryQueue;
|
||||
let _pool = RateLimitedWorkerPool::new(
|
||||
InMemoryQueue::new(),
|
||||
WorkerConfig::default(),
|
||||
10.0,
|
||||
5,
|
||||
);
|
||||
let _pool =
|
||||
RateLimitedWorkerPool::new(InMemoryQueue::new(), WorkerConfig::default(), 10.0, 5);
|
||||
// smoke: just verifies construction
|
||||
}
|
||||
}
|
||||
|
||||
@@ -36,7 +36,11 @@ async fn rate_limited_queue_no_item_loss_under_contention() {
|
||||
let s = Arc::clone(&seen);
|
||||
workers.push(tokio::spawn(async move {
|
||||
loop {
|
||||
match q.dequeue("stress", Duration::from_millis(50)).await.unwrap() {
|
||||
match q
|
||||
.dequeue("stress", Duration::from_millis(50))
|
||||
.await
|
||||
.unwrap()
|
||||
{
|
||||
Some(job) => {
|
||||
let _ = String::from_utf8(job.payload).unwrap();
|
||||
s.fetch_add(1, Ordering::SeqCst);
|
||||
@@ -64,4 +68,4 @@ async fn rate_limited_queue_no_item_loss_under_contention() {
|
||||
TASKS * PER_TASK,
|
||||
"items lost under contention"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-storage"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -67,12 +67,10 @@ impl StorageDriver for LocalFileStorage {
|
||||
let mut file = tokio::fs::File::create(&tmp)
|
||||
.await
|
||||
.map_err(|e| StorageError::Io(e.to_string()))?;
|
||||
let written = tokio::io::copy(&mut data, &mut file)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
let _ = std::fs::remove_file(&tmp);
|
||||
StorageError::Io(e.to_string())
|
||||
})?;
|
||||
let written = tokio::io::copy(&mut data, &mut file).await.map_err(|e| {
|
||||
let _ = std::fs::remove_file(&tmp);
|
||||
StorageError::Io(e.to_string())
|
||||
})?;
|
||||
// Ensure durability: flush to OS, fsync, then rename.
|
||||
tokio::fs::File::open(&tmp)
|
||||
.await
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse-tracing"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
@@ -14,9 +14,7 @@ impl TracingLayer {
|
||||
pub fn install() {
|
||||
let filter = EnvFilter::try_from_default_env()
|
||||
.unwrap_or_else(|_| EnvFilter::new("mytheclipse=info"));
|
||||
let _ = tracing_subscriber::fmt()
|
||||
.with_env_filter(filter)
|
||||
.try_init();
|
||||
let _ = tracing_subscriber::fmt().with_env_filter(filter).try_init();
|
||||
}
|
||||
|
||||
/// Returns a formatted layer for manual composition.
|
||||
|
||||
@@ -3,9 +3,21 @@
|
||||
//! Pre-built tracing layers combining all mytheclipse primitives with
|
||||
//! optional export backends (OTLP, Jaeger, Zipkin).
|
||||
|
||||
#[cfg(any(
|
||||
feature = "env",
|
||||
feature = "otel",
|
||||
feature = "jaeger",
|
||||
feature = "zipkin"
|
||||
))]
|
||||
pub mod fmt;
|
||||
pub mod otel;
|
||||
|
||||
#[cfg(any(
|
||||
feature = "env",
|
||||
feature = "otel",
|
||||
feature = "jaeger",
|
||||
feature = "zipkin"
|
||||
))]
|
||||
pub use fmt::TracingLayer;
|
||||
#[cfg(any(feature = "otel", feature = "jaeger", feature = "full"))]
|
||||
pub use otel::OtelLayer;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "mytheclipse"
|
||||
version = "1.21.0"
|
||||
version = "1.21.2"
|
||||
edition = "2021"
|
||||
rust-version = "1.75"
|
||||
license = "MIT OR Apache-2.0"
|
||||
@@ -43,11 +43,21 @@ name = "main"
|
||||
path = "examples/main.rs"
|
||||
required-features = ["full"]
|
||||
|
||||
[[example]]
|
||||
name = "high_level"
|
||||
path = "examples/high_level.rs"
|
||||
required-features = ["full"]
|
||||
|
||||
[[example]]
|
||||
name = "scaling_demo"
|
||||
path = "examples/scaling_demo.rs"
|
||||
required-features = ["full"]
|
||||
|
||||
[[test]]
|
||||
name = "race_stress"
|
||||
path = "tests/race_stress.rs"
|
||||
required-features = ["full"]
|
||||
|
||||
[[bench]]
|
||||
name = "primitives"
|
||||
path = "benches/primitives.rs"
|
||||
|
||||
@@ -18,8 +18,8 @@ use mytheclipse::aggregate_error::AggregateError;
|
||||
use mytheclipse::parallel_map::{parallel_for_each, parallel_map};
|
||||
use mytheclipse::pool::{Pool, SemaphorePool};
|
||||
use mytheclipse::ratelimit::RateLimiter;
|
||||
use mytheclipse::retry_ext::RetryExt;
|
||||
use mytheclipse::retry::RetryConfig;
|
||||
use mytheclipse::retry_ext::RetryExt;
|
||||
use mytheclipse::shutdown_guard::ShutdownGuard;
|
||||
|
||||
fn rt() -> Runtime {
|
||||
@@ -52,9 +52,13 @@ fn bench_parallel_for_each(c: &mut Criterion) {
|
||||
let rt = rt();
|
||||
c.bench_function("parallel_for_each/1000x8", |b| {
|
||||
b.to_async(&rt).iter(|| async {
|
||||
parallel_for_each(0u32..1000, 8, |_| async move { Ok::<_, std::io::Error>(()) })
|
||||
.await
|
||||
.unwrap();
|
||||
parallel_for_each(
|
||||
0u32..1000,
|
||||
8,
|
||||
|_| async move { Ok::<_, std::io::Error>(()) },
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
black_box(());
|
||||
});
|
||||
});
|
||||
@@ -177,4 +181,4 @@ criterion_group!(
|
||||
bench_aggregate_error,
|
||||
bench_shutdown_guard,
|
||||
);
|
||||
criterion_main!(benches);
|
||||
criterion_main!(benches);
|
||||
|
||||
@@ -21,8 +21,8 @@ use mytheclipse::{
|
||||
pool::{AutoReconnectPool, Pool, Reconnectable, SemaphorePool},
|
||||
retry_ext::RetryExt,
|
||||
runtime_auto::RuntimeConfig,
|
||||
shutdown_guard::ShutdownGuard,
|
||||
service_builder::ServiceConfig,
|
||||
shutdown_guard::ShutdownGuard,
|
||||
};
|
||||
|
||||
#[tokio::main]
|
||||
@@ -66,7 +66,10 @@ async fn main() {
|
||||
}
|
||||
};
|
||||
let value = fut.retry(cfg, |_: &String| true, op).await.unwrap();
|
||||
println!("3. RetryExt with {} attempts -> {value}", attempts.load(Ordering::SeqCst));
|
||||
println!(
|
||||
"3. RetryExt with {} attempts -> {value}",
|
||||
attempts.load(Ordering::SeqCst)
|
||||
);
|
||||
|
||||
// 4. RAII ShutdownGuard — callback runs exactly once on drop, panic-safe.
|
||||
let fired = Arc::new(AtomicU32::new(0));
|
||||
@@ -93,18 +96,20 @@ async fn main() {
|
||||
let svc = AutoMetricsServiceBuilder::new("demo_op", svc_cfg);
|
||||
let n = Arc::new(AtomicU32::new(0));
|
||||
let n2 = Arc::clone(&n);
|
||||
let _: Result<u32, mytheclipse::service_builder::RunError<()>> = svc.run(|| {
|
||||
let n2 = Arc::clone(&n2);
|
||||
Box::pin(async move {
|
||||
tokio::time::sleep(Duration::from_millis(5)).await;
|
||||
let v = n2.fetch_add(1, Ordering::SeqCst);
|
||||
if v < 1 {
|
||||
Err(())
|
||||
} else {
|
||||
Ok(7u32)
|
||||
}
|
||||
let _: Result<u32, mytheclipse::service_builder::RunError<()>> = svc
|
||||
.run(|| {
|
||||
let n2 = Arc::clone(&n2);
|
||||
Box::pin(async move {
|
||||
tokio::time::sleep(Duration::from_millis(5)).await;
|
||||
let v = n2.fetch_add(1, Ordering::SeqCst);
|
||||
if v < 1 {
|
||||
Err(())
|
||||
} else {
|
||||
Ok(7u32)
|
||||
}
|
||||
})
|
||||
})
|
||||
}).await;
|
||||
.await;
|
||||
let snap = svc.collector().snapshot();
|
||||
println!(
|
||||
"6. AutoMetrics -> {} counters, {} histograms",
|
||||
|
||||
@@ -16,7 +16,10 @@ use mytheclipse::parallel_map::{parallel_for_each, ParallelConcurrency};
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
let total = 100u32;
|
||||
println!("host available_parallelism = {}", <() as ParallelConcurrency>::resolve(()));
|
||||
println!(
|
||||
"host available_parallelism = {}",
|
||||
<() as ParallelConcurrency>::resolve(())
|
||||
);
|
||||
println!("total items = {total}");
|
||||
println!();
|
||||
|
||||
@@ -49,6 +52,12 @@ async fn run(label: &str, concurrency: impl ParallelConcurrency + Copy, total: u
|
||||
let elapsed = start.elapsed();
|
||||
|
||||
println!("{label} resolved={resolved}");
|
||||
println!(" peak in-flight = {} (bounded, never {total})", peak.load(Ordering::SeqCst));
|
||||
println!(" elapsed = {elapsed:?} (sequential ~{}ms)", total * 1);
|
||||
}
|
||||
println!(
|
||||
" peak in-flight = {} (bounded, never {total})",
|
||||
peak.load(Ordering::SeqCst)
|
||||
);
|
||||
println!(
|
||||
" elapsed = {elapsed:?} (sequential ~{}ms)",
|
||||
total * 1
|
||||
);
|
||||
}
|
||||
|
||||
@@ -117,12 +117,18 @@ impl std::error::Error for AggregateError {}
|
||||
|
||||
impl From<Vec<Box<dyn std::error::Error + Send + Sync>>> for AggregateError {
|
||||
fn from(errors: Vec<Box<dyn std::error::Error + Send + Sync>>) -> Self {
|
||||
Self { errors, context: None }
|
||||
Self {
|
||||
errors,
|
||||
context: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Extend<Box<dyn std::error::Error + Send + Sync>> for AggregateError {
|
||||
fn extend<T: IntoIterator<Item = Box<dyn std::error::Error + Send + Sync>>>(&mut self, iter: T) {
|
||||
fn extend<T: IntoIterator<Item = Box<dyn std::error::Error + Send + Sync>>>(
|
||||
&mut self,
|
||||
iter: T,
|
||||
) {
|
||||
self.errors.extend(iter);
|
||||
}
|
||||
}
|
||||
@@ -160,8 +166,7 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn collects_values_when_all_ok() {
|
||||
let results: Vec<Result<u32, std::io::Error>> =
|
||||
vec![Ok(1), Ok(2), Ok(3)];
|
||||
let results: Vec<Result<u32, std::io::Error>> = vec![Ok(1), Ok(2), Ok(3)];
|
||||
let out = AggregateError::from_results(results).unwrap();
|
||||
assert_eq!(out, vec![1, 2, 3]);
|
||||
}
|
||||
|
||||
@@ -74,8 +74,8 @@ impl AutoMetricsServiceBuilder {
|
||||
self
|
||||
}
|
||||
|
||||
/// Attaches a [`MetricsBridge`] to forward snapshots downstream (requires
|
||||
/// the `resiliency` feature which pulls in the bridge).
|
||||
/// Attaches a [`crate::metrics_bridge::MetricsBridge`] to forward snapshots
|
||||
/// downstream (requires the `resiliency` feature which pulls in the bridge).
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub fn with_bridge(mut self, bridge: crate::metrics_bridge::MetricsBridge) -> Self {
|
||||
self.bridge = Some(bridge);
|
||||
@@ -108,8 +108,13 @@ impl AutoMetricsServiceBuilder {
|
||||
Err(_) => "other",
|
||||
};
|
||||
|
||||
self.metrics
|
||||
.inc_counter(&format!("mytheclipse_service_calls_total{{service=\"{}\",outcome=\"{}\"}}", self.service_name, outcome), 1);
|
||||
self.metrics.inc_counter(
|
||||
&format!(
|
||||
"mytheclipse_service_calls_total{{service=\"{}\",outcome=\"{}\"}}",
|
||||
self.service_name, outcome
|
||||
),
|
||||
1,
|
||||
);
|
||||
self.metrics
|
||||
.observe("mytheclipse_service_duration_seconds", dur);
|
||||
|
||||
@@ -125,8 +130,8 @@ impl AutoMetricsServiceBuilder {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{Ordering, AtomicU32};
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_metrics_records_call() {
|
||||
@@ -136,13 +141,19 @@ mod tests {
|
||||
|
||||
let attempts = Arc::new(AtomicU32::new(0));
|
||||
let a = Arc::clone(&attempts);
|
||||
let result: Result<u32, RunError<()>> = builder.run(|| {
|
||||
let a = Arc::clone(&a);
|
||||
Box::pin(async move {
|
||||
let n = a.fetch_add(1, Ordering::SeqCst);
|
||||
if n < 2 { Err(()) } else { Ok(42u32) }
|
||||
let result: Result<u32, RunError<()>> = builder
|
||||
.run(|| {
|
||||
let a = Arc::clone(&a);
|
||||
Box::pin(async move {
|
||||
let n = a.fetch_add(1, Ordering::SeqCst);
|
||||
if n < 2 {
|
||||
Err(())
|
||||
} else {
|
||||
Ok(42u32)
|
||||
}
|
||||
})
|
||||
})
|
||||
}).await;
|
||||
.await;
|
||||
assert_eq!(result.unwrap(), 42);
|
||||
assert_eq!(attempts.load(Ordering::SeqCst), 3);
|
||||
let snap = builder.collector().snapshot();
|
||||
|
||||
@@ -8,7 +8,6 @@
|
||||
//! *concurrency*, `BgJoiner` adds structured *lifetimes* so a service can wait
|
||||
//! for all in-flight work to settle before terminating.
|
||||
|
||||
use std::future::Future;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
@@ -36,7 +35,9 @@ impl BgJoiner {
|
||||
F: std::future::Future + Send + 'static,
|
||||
F::Output: Send + 'static,
|
||||
{
|
||||
let handle: JoinHandle<()> = tokio::spawn(async move { let _ = future.await; });
|
||||
let handle: JoinHandle<()> = tokio::spawn(async move {
|
||||
let _ = future.await;
|
||||
});
|
||||
self.track(handle);
|
||||
}
|
||||
|
||||
@@ -55,6 +56,11 @@ impl BgJoiner {
|
||||
self.inner.lock().await.len()
|
||||
}
|
||||
|
||||
/// Returns `true` if there are no currently-tracked tasks.
|
||||
pub async fn is_empty(&self) -> bool {
|
||||
self.inner.lock().await.is_empty()
|
||||
}
|
||||
|
||||
/// Await every tracked task, dropping any that are still pending once
|
||||
/// `deadline` elapses. Returns the count of tasks that had not completed
|
||||
/// within the timeout.
|
||||
@@ -66,7 +72,6 @@ impl BgJoiner {
|
||||
};
|
||||
|
||||
let mut pending: Vec<JoinHandle<()>> = handles;
|
||||
let mut dropped = 0usize;
|
||||
|
||||
loop {
|
||||
if pending.is_empty() {
|
||||
@@ -74,11 +79,11 @@ impl BgJoiner {
|
||||
}
|
||||
|
||||
if now.elapsed() >= deadline {
|
||||
dropped = pending.len();
|
||||
let count = pending.len();
|
||||
for h in pending.drain(..) {
|
||||
h.abort();
|
||||
}
|
||||
return dropped;
|
||||
return count;
|
||||
}
|
||||
|
||||
let remaining = deadline.saturating_sub(now.elapsed());
|
||||
|
||||
@@ -121,18 +121,16 @@ where
|
||||
F: Fn(I::Item) -> Result<T, String> + Send + Sync,
|
||||
{
|
||||
let wrapped = AssertUnwindSafe(f);
|
||||
let collected: Vec<Result<T, String>> = context()
|
||||
.compute_pool
|
||||
.install(move || {
|
||||
let f = wrapped;
|
||||
items
|
||||
.into_par_iter()
|
||||
.map(|item| {
|
||||
catch_unwind(AssertUnwindSafe(|| f(item)))
|
||||
.unwrap_or_else(|payload| Err(panic_payload_to_string(payload)))
|
||||
})
|
||||
.collect()
|
||||
});
|
||||
let collected: Vec<Result<T, String>> = context().compute_pool.install(move || {
|
||||
let f = wrapped;
|
||||
items
|
||||
.into_par_iter()
|
||||
.map(|item| {
|
||||
catch_unwind(AssertUnwindSafe(|| f(item)))
|
||||
.unwrap_or_else(|payload| Err(panic_payload_to_string(payload)))
|
||||
})
|
||||
.collect()
|
||||
});
|
||||
|
||||
let mut values = Vec::with_capacity(collected.len());
|
||||
let mut errors = Vec::new();
|
||||
@@ -165,10 +163,7 @@ where
|
||||
/// ).unwrap();
|
||||
/// assert_eq!(a + b, (0..2_000_000u64).sum::<u64>());
|
||||
/// ```
|
||||
pub fn compute_join<A, RA, B, RB>(
|
||||
a: A,
|
||||
b: B,
|
||||
) -> Result<(RA, RB), MytheclipseError>
|
||||
pub fn compute_join<A, RA, B, RB>(a: A, b: B) -> Result<(RA, RB), MytheclipseError>
|
||||
where
|
||||
A: FnOnce() -> RA + Send,
|
||||
RA: Send,
|
||||
@@ -177,15 +172,19 @@ where
|
||||
{
|
||||
let a = AssertUnwindSafe(a);
|
||||
let b = AssertUnwindSafe(b);
|
||||
context()
|
||||
.compute_pool
|
||||
.install(|| {
|
||||
let (ra, rb) = rayon::join(
|
||||
move || catch_unwind(a).map_err(|p| MytheclipseError::ComputePanic(panic_payload_to_string(p))),
|
||||
move || catch_unwind(b).map_err(|p| MytheclipseError::ComputePanic(panic_payload_to_string(p))),
|
||||
);
|
||||
Ok((ra?, rb?))
|
||||
})
|
||||
context().compute_pool.install(|| {
|
||||
let (ra, rb) = rayon::join(
|
||||
move || {
|
||||
catch_unwind(a)
|
||||
.map_err(|p| MytheclipseError::ComputePanic(panic_payload_to_string(p)))
|
||||
},
|
||||
move || {
|
||||
catch_unwind(b)
|
||||
.map_err(|p| MytheclipseError::ComputePanic(panic_payload_to_string(p)))
|
||||
},
|
||||
);
|
||||
Ok((ra?, rb?))
|
||||
})
|
||||
}
|
||||
|
||||
/// Runs `f` over every item on the compute pool in parallel, discarding
|
||||
@@ -211,25 +210,20 @@ where
|
||||
F: Fn(I::Item) -> Result<(), String> + Send + Sync,
|
||||
{
|
||||
let wrapped = AssertUnwindSafe(f);
|
||||
let collected: Vec<Result<(), String>> = context()
|
||||
.compute_pool
|
||||
.install(move || {
|
||||
let f = wrapped;
|
||||
items
|
||||
.into_par_iter()
|
||||
.map(|item| {
|
||||
catch_unwind(AssertUnwindSafe(|| f(item)))
|
||||
.unwrap_or_else(|payload| Err(panic_payload_to_string(payload)))
|
||||
})
|
||||
.collect()
|
||||
});
|
||||
let collected: Vec<Result<(), String>> = context().compute_pool.install(move || {
|
||||
let f = wrapped;
|
||||
items
|
||||
.into_par_iter()
|
||||
.map(|item| {
|
||||
catch_unwind(AssertUnwindSafe(|| f(item)))
|
||||
.unwrap_or_else(|payload| Err(panic_payload_to_string(payload)))
|
||||
})
|
||||
.collect()
|
||||
});
|
||||
|
||||
if collected.iter().any(|r| r.is_err()) {
|
||||
Err(ComputeErrors {
|
||||
errors: collected
|
||||
.into_iter()
|
||||
.filter_map(|r| r.err())
|
||||
.collect(),
|
||||
errors: collected.into_iter().filter_map(|r| r.err()).collect(),
|
||||
})
|
||||
} else {
|
||||
Ok(())
|
||||
|
||||
@@ -69,7 +69,12 @@ impl Drop for LockGuard {
|
||||
#[async_trait]
|
||||
pub trait DistributedLock: Send + Sync {
|
||||
/// Attempts to acquire the lock with the given lease duration.
|
||||
async fn acquire(&self, key: &str, lease: Duration, timeout: Duration) -> Result<LockGuard, LockError>;
|
||||
async fn acquire(
|
||||
&self,
|
||||
key: &str,
|
||||
lease: Duration,
|
||||
timeout: Duration,
|
||||
) -> Result<LockGuard, LockError>;
|
||||
|
||||
/// Releases the lock.
|
||||
async fn release(&self, key: &str) -> Result<(), LockError>;
|
||||
@@ -90,14 +95,6 @@ impl InProcLock {
|
||||
held: Arc::new(Mutex::new(std::collections::HashMap::new())),
|
||||
}
|
||||
}
|
||||
|
||||
fn is_expired(map: &std::collections::HashMap<String, Instant>, key: &str) -> bool {
|
||||
if let Some(expiry) = map.get(key) {
|
||||
*expiry <= Instant::now()
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for InProcLock {
|
||||
@@ -108,7 +105,12 @@ impl Default for InProcLock {
|
||||
|
||||
#[async_trait]
|
||||
impl DistributedLock for InProcLock {
|
||||
async fn acquire(&self, key: &str, lease: Duration, timeout: Duration) -> Result<LockGuard, LockError> {
|
||||
async fn acquire(
|
||||
&self,
|
||||
key: &str,
|
||||
lease: Duration,
|
||||
timeout: Duration,
|
||||
) -> Result<LockGuard, LockError> {
|
||||
let deadline = Instant::now() + timeout;
|
||||
loop {
|
||||
{
|
||||
@@ -154,7 +156,10 @@ mod tests {
|
||||
#[tokio::test]
|
||||
async fn lock_acquire_release() {
|
||||
let lock = InProcLock::new();
|
||||
let guard = lock.acquire("key", Duration::from_secs(10), Duration::from_secs(1)).await.unwrap();
|
||||
let guard = lock
|
||||
.acquire("key", Duration::from_secs(10), Duration::from_secs(1))
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(lock.release("key").await.is_ok());
|
||||
drop(guard);
|
||||
}
|
||||
@@ -162,9 +167,14 @@ mod tests {
|
||||
#[tokio::test]
|
||||
async fn lock_rejects_second_acquire() {
|
||||
let lock = InProcLock::new();
|
||||
let _guard1 = lock.acquire("key", Duration::from_secs(10), Duration::from_secs(1)).await.unwrap();
|
||||
let _guard1 = lock
|
||||
.acquire("key", Duration::from_secs(10), Duration::from_secs(1))
|
||||
.await
|
||||
.unwrap();
|
||||
// While guard1 is alive, a second acquire with short timeout should fail.
|
||||
let result = lock.acquire("key", Duration::from_secs(10), Duration::from_millis(50)).await;
|
||||
let result = lock
|
||||
.acquire("key", Duration::from_secs(10), Duration::from_millis(50))
|
||||
.await;
|
||||
assert!(result.is_err());
|
||||
drop(_guard1);
|
||||
}
|
||||
@@ -172,21 +182,31 @@ mod tests {
|
||||
#[tokio::test]
|
||||
async fn lock_auto_releases_on_drop() {
|
||||
let lock = InProcLock::new();
|
||||
let guard = lock.acquire("k", Duration::from_secs(10), Duration::from_secs(1)).await.unwrap();
|
||||
let guard = lock
|
||||
.acquire("k", Duration::from_secs(10), Duration::from_secs(1))
|
||||
.await
|
||||
.unwrap();
|
||||
drop(guard);
|
||||
// After drop, the lock should be releasable / re-acquirable.
|
||||
let result = lock.acquire("k", Duration::from_secs(10), Duration::from_millis(50)).await;
|
||||
let result = lock
|
||||
.acquire("k", Duration::from_secs(10), Duration::from_millis(50))
|
||||
.await;
|
||||
assert!(result.is_ok(), "lock should be free after guard drop");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn lock_expires_after_lease() {
|
||||
let lock = InProcLock::new();
|
||||
let _guard = lock.acquire("key", Duration::from_millis(20), Duration::from_millis(5)).await.unwrap();
|
||||
let _guard = lock
|
||||
.acquire("key", Duration::from_millis(20), Duration::from_millis(5))
|
||||
.await
|
||||
.unwrap();
|
||||
drop(_guard);
|
||||
tokio::time::sleep(Duration::from_millis(30)).await;
|
||||
// Should be acquirable now.
|
||||
let result = lock.acquire("key", Duration::from_millis(20), Duration::from_millis(5)).await;
|
||||
let result = lock
|
||||
.acquire("key", Duration::from_millis(20), Duration::from_millis(5))
|
||||
.await;
|
||||
assert!(result.is_ok());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -26,7 +26,9 @@ impl fmt::Display for HealthStatus {
|
||||
/// A single health check.
|
||||
pub trait HealthCheck: Send + Sync {
|
||||
fn name(&self) -> &str;
|
||||
fn check(&self) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>>;
|
||||
fn check(
|
||||
&self,
|
||||
) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>>;
|
||||
}
|
||||
|
||||
/// A registered health check with its name and trait object.
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
//! - [`spawn_io`] (feature `io`) — spawn an async I/O task, tracing-instrumented.
|
||||
//! - [`compute()`] (feature `compute`) — run CPU-bound work on a sized Rayon pool, panic-isolated.
|
||||
//! - [`spawn_bg`] (feature `bg`) — spawn a background task under bounded concurrency.
|
||||
//! - [`retry`] / [`CircuitBreaker`] / [`timeout()`] (feature `resiliency`) — fault tolerance.
|
||||
//! - [`retry()`] / [`CircuitBreaker`] / [`timeout()`] (feature `resiliency`) — fault tolerance.
|
||||
//! - [`RateLimiter`] / [`BackpressureQueue`] / [`ConcurrencyLimiter`] (feature `traffic`) — load control.
|
||||
//! - [`SemaphorePool`] (feature `traffic`) — shared bounded resource pool.
|
||||
//! - [`ShutdownManager`] / [`CronSchedule`] (feature `lifecycle`) — lifecycle + scheduling.
|
||||
@@ -24,57 +24,59 @@
|
||||
pub mod context;
|
||||
pub mod error;
|
||||
|
||||
#[cfg(feature = "io")]
|
||||
pub mod io;
|
||||
#[cfg(feature = "compute")]
|
||||
pub mod compute;
|
||||
#[cfg(feature = "bg")]
|
||||
pub mod bg;
|
||||
#[cfg(feature = "compute")]
|
||||
pub mod compute;
|
||||
#[cfg(feature = "io")]
|
||||
pub mod io;
|
||||
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod retry;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod retry_ext;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod aggregate_error;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod parallel_map;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use retry_ext::RetryExt;
|
||||
pub mod retry;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod retry_ext;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use aggregate_error::AggregateError;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use parallel_map::{parallel_map, parallel_map_unordered, parallel_for_each, ParallelConcurrency};
|
||||
#[cfg(feature = "observability")]
|
||||
pub use parallel_map::{
|
||||
parallel_for_each, parallel_map, parallel_map_unordered, ParallelConcurrency,
|
||||
};
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use retry_ext::RetryExt;
|
||||
#[cfg(all(feature = "observability", feature = "resiliency"))]
|
||||
pub mod auto_metrics_service;
|
||||
#[cfg(feature = "observability")]
|
||||
#[cfg(all(feature = "observability", feature = "resiliency"))]
|
||||
pub use auto_metrics_service::AutoMetricsServiceBuilder;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod circuit_breaker;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod timeout;
|
||||
|
||||
#[cfg(feature = "traffic")]
|
||||
pub mod ratelimit;
|
||||
#[cfg(feature = "traffic")]
|
||||
pub mod backpressure;
|
||||
#[cfg(feature = "traffic")]
|
||||
pub mod concurrency;
|
||||
#[cfg(feature = "traffic")]
|
||||
pub mod pool;
|
||||
#[cfg(feature = "traffic")]
|
||||
pub mod ratelimit;
|
||||
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub mod shutdown;
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub mod cron;
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub mod health;
|
||||
#[cfg(all(feature = "observability", feature = "traffic"))]
|
||||
pub mod pool_health;
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub mod leader;
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub mod lifecycle;
|
||||
#[cfg(all(feature = "observability", feature = "traffic"))]
|
||||
pub mod pool_health;
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub mod shutdown;
|
||||
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub mod bg_join;
|
||||
@@ -94,59 +96,59 @@ pub mod middleware;
|
||||
#[cfg(feature = "observability")]
|
||||
pub mod metrics;
|
||||
#[cfg(feature = "observability")]
|
||||
pub mod panic_tracker;
|
||||
#[cfg(feature = "observability")]
|
||||
pub mod metrics_bridge;
|
||||
#[cfg(feature = "observability")]
|
||||
pub mod panic_tracker;
|
||||
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod service_builder;
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub mod dlock;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod service_builder;
|
||||
|
||||
pub use context::{context, EngineContext};
|
||||
pub use error::MytheclipseError;
|
||||
|
||||
#[cfg(feature = "io")]
|
||||
pub use io::spawn_io;
|
||||
#[cfg(feature = "compute")]
|
||||
pub use compute::{compute, compute_join, compute_map, compute_par_for_each, ComputeErrors};
|
||||
#[cfg(feature = "bg")]
|
||||
pub use bg::spawn_bg;
|
||||
#[cfg(feature = "compute")]
|
||||
pub use compute::{compute, compute_join, compute_map, compute_par_for_each, ComputeErrors};
|
||||
#[cfg(feature = "io")]
|
||||
pub use io::spawn_io;
|
||||
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use retry::{retry, JitterKind, RetryConfig, RetryError};
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitError, CircuitState};
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use retry::{retry, JitterKind, RetryConfig, RetryError};
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use timeout::{timeout, with_timeout, Timeout, TimeoutError};
|
||||
|
||||
#[cfg(feature = "traffic")]
|
||||
pub use ratelimit::{RateLimitError, RateLimiter};
|
||||
/// Re-export of `async-trait` so implementing [`pool::Reconnectable`] (and
|
||||
/// other async traits) doesn't require users to add their own `async-trait`
|
||||
/// dependency.
|
||||
pub use async_trait::async_trait;
|
||||
#[cfg(feature = "traffic")]
|
||||
pub use backpressure::{BackpressureError, BackpressureQueue, OverflowPolicy};
|
||||
#[cfg(feature = "traffic")]
|
||||
pub use concurrency::{ConcurrencyLimiter, ConcurrencyPermit};
|
||||
#[cfg(feature = "traffic")]
|
||||
pub use pool::{Pool, PoolError, Pooled, SemaphorePool, AutoReconnectPool, Reconnectable};
|
||||
/// Re-export of `async-trait` so implementing [`pool::Reconnectable`] (and
|
||||
/// other async traits) doesn't require users to add their own `async-trait`
|
||||
/// dependency.
|
||||
pub use async_trait::async_trait;
|
||||
pub use pool::{AutoReconnectPool, Pool, PoolError, Pooled, Reconnectable, SemaphorePool};
|
||||
#[cfg(feature = "traffic")]
|
||||
pub use ratelimit::{RateLimitError, RateLimiter};
|
||||
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub use shutdown::{ShutdownManager, ShutdownSignal};
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub use cron::{schedule, CronError, CronJob, CronParseError, CronSchedule};
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub use health::{HealthCheck, HealthRegistry, HealthStatus};
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub use leader::{InProcLeaderElection, LeaderElection};
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub use shutdown::{ShutdownManager, ShutdownSignal};
|
||||
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use service_builder::{RunError, ServiceBuilder, ServiceConfig};
|
||||
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub use dlock::{DistributedLock, LockError, LockGuard, InProcLock};
|
||||
pub use dlock::{DistributedLock, InProcLock, LockError, LockGuard};
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub use lifecycle::AsyncLifecycleManager;
|
||||
|
||||
@@ -169,7 +171,7 @@ pub use pool_health::HealthCheckedPool;
|
||||
pub use bg_join::BgJoiner;
|
||||
|
||||
#[cfg(all(feature = "observability", feature = "resiliency"))]
|
||||
pub use middleware::{MiddlewarePipeline, PipelineError, BoxMiddleware, mw};
|
||||
pub use middleware::{mw, BoxMiddleware, MiddlewarePipeline, PipelineError};
|
||||
#[cfg(feature = "observability")]
|
||||
pub use panic_tracker::{PanicGuard, PanicInfo, PanicTracker};
|
||||
|
||||
|
||||
@@ -15,9 +15,7 @@ use crate::shutdown::ShutdownManager;
|
||||
///
|
||||
/// Typical usage:
|
||||
/// ```ignore
|
||||
/// # tokio::runtime::Runtime::new().unwrap().block_on(async {
|
||||
/// # use mytheclipse::AsyncLifecycleManager;
|
||||
/// let mgr = AsyncLifecycleManager::new();
|
||||
/// let mgr = mytheclipse::AsyncLifecycleManager::new();
|
||||
/// mgr.register_health_check("db", my_db_check());
|
||||
/// let handle = mgr.start_health_loop(std::time::Duration::from_secs(30));
|
||||
/// mgr.await_shutdown(std::time::Duration::from_secs(10)).await;
|
||||
@@ -47,7 +45,11 @@ impl AsyncLifecycleManager {
|
||||
}
|
||||
|
||||
/// Registers a named health check.
|
||||
pub async fn register_health_check(&self, name: impl Into<String>, check: impl HealthCheck + 'static) {
|
||||
pub async fn register_health_check(
|
||||
&self,
|
||||
name: impl Into<String>,
|
||||
check: impl HealthCheck + 'static,
|
||||
) {
|
||||
self.health.register(name, check).await;
|
||||
}
|
||||
|
||||
@@ -73,7 +75,7 @@ impl AsyncLifecycleManager {
|
||||
loop {
|
||||
// Stop when shutdown is requested.
|
||||
if sig.is_shutdown() {
|
||||
tracing::info_span!("mytheclipse_health_loop", );
|
||||
tracing::info_span!("mytheclipse_health_loop",);
|
||||
return;
|
||||
}
|
||||
tokio::select! {
|
||||
@@ -124,16 +126,26 @@ mod tests {
|
||||
|
||||
struct AlwaysOk;
|
||||
impl HealthCheck for AlwaysOk {
|
||||
fn name(&self) -> &str { "always-ok" }
|
||||
fn check(&self) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>> {
|
||||
fn name(&self) -> &str {
|
||||
"always-ok"
|
||||
}
|
||||
fn check(
|
||||
&self,
|
||||
) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>>
|
||||
{
|
||||
Box::pin(async { HealthStatus::Ok })
|
||||
}
|
||||
}
|
||||
|
||||
struct AlwaysBad;
|
||||
impl HealthCheck for AlwaysBad {
|
||||
fn name(&self) -> &str { "always-bad" }
|
||||
fn check(&self) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>> {
|
||||
fn name(&self) -> &str {
|
||||
"always-bad"
|
||||
}
|
||||
fn check(
|
||||
&self,
|
||||
) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>>
|
||||
{
|
||||
Box::pin(async { HealthStatus::Unhealthy })
|
||||
}
|
||||
}
|
||||
|
||||
Some files were not shown because too many files have changed in this diff Show More
Reference in New Issue
Block a user