Compare commits

..
8 Commits
Author SHA1 Message Date
semantic-release-bot 3376bee1d3 chore(release): 1.17.0 [skip ci]
# [1.17.0](https://github.com/asepharyana/mytheclipse/compare/v1.16.0...v1.17.0) (2026-08-29)

### Features

* round-14 abstractions — parallel_map bounded fan-out ([bb4998d](https://github.com/asepharyana/mytheclipse/commit/bb4998d8fa68e25415ea79a245313f17a2792a06))
2026-08-29 14:00:53 +00:00
asepharyana bb4998d8fa feat: round-14 abstractions — parallel_map bounded fan-out
CI / Rustfmt (push) Canceled after 0s
CI / Clippy (push) Canceled after 0s
CI / Test (workspace all features) (push) Canceled after 0s
CI / Test (workspace default features) (push) Canceled after 0s
CI / Test (mytheclipse / bg only) (push) Canceled after 0s
CI / Test (mytheclipse / compute only) (push) Canceled after 0s
CI / Test (mytheclipse / io only) (push) Canceled after 0s
CI / Test (mytheclipse / lifecycle only) (push) Canceled after 0s
CI / Test (mytheclipse / observability only) (push) Canceled after 0s
CI / Test (mytheclipse / resiliency only) (push) Canceled after 0s
CI / Test (mytheclipse / traffic only) (push) Canceled after 0s
CI / Test (mytheclipse-cache / l2-redis) (push) Canceled after 0s
CI / Test (mytheclipse-cache / l1-moka) (push) Canceled after 0s
CI / Test (mytheclipse-cache / default) (push) Canceled after 0s
CI / Test (mytheclipse-config / default) (push) Canceled after 0s
CI / Test (mytheclipse-crypto / default) (push) Canceled after 0s
CI / Test (mytheclipse-event / amqp) (push) Canceled after 0s
CI / Test (mytheclipse-event / nats) (push) Canceled after 0s
CI / Test (mytheclipse-event / default (mem)) (push) Canceled after 0s
CI / Test (mytheclipse-storage / gcs) (push) Canceled after 0s
CI / Test (mytheclipse-storage / s3) (push) Canceled after 0s
CI / Test (mytheclipse-storage / default (local)) (push) Canceled after 0s
CI / Run mytheclipse example (push) Canceled after 0s
CI / Docs check (push) Canceled after 0s
CI / Cargo package dry-run (mytheclipse) (push) Canceled after 0s
CI / Cargo package dry-run (mytheclipse-cache) (push) Canceled after 0s
CI / Cargo package dry-run (mytheclipse-config) (push) Canceled after 0s
CI / Cargo package dry-run (mytheclipse-crypto) (push) Canceled after 0s
CI / Cargo package dry-run (mytheclipse-event) (push) Canceled after 0s
CI / Cargo package dry-run (mytheclipse-storage) (push) Canceled after 0s
Release / Semantic Release (push) Canceled after 0s
2026-08-29 20:59:49 +07:00
semantic-release-bot 125bc57290 chore(release): 1.16.0 [skip ci]
# [1.16.0](https://github.com/asepharyana/mytheclipse/compare/v1.15.0...v1.16.0) (2026-08-29)

### Features

* round-13 abstractions — AggregateError for parallel fan-out ([ff5fbe4](https://github.com/asepharyana/mytheclipse/commit/ff5fbe49dc1ff06c287306835110a6652d6b24b9))
2026-08-29 13:56:03 +00:00
asepharyana ff5fbe49dc feat: round-13 abstractions — AggregateError for parallel fan-out 2026-08-29 20:55:07 +07:00
semantic-release-bot 8d4d8aa52c chore(release): 1.15.0 [skip ci]
# [1.15.0](https://github.com/asepharyana/mytheclipse/compare/v1.14.0...v1.15.0) (2026-08-29)

### Features

* round-12 abstractions — AutoReconnectPool, Reconnectable ([ddb2c2f](https://github.com/asepharyana/mytheclipse/commit/ddb2c2fd2d43e07b0d26b2901746ffcc3fe8b284))
2026-08-29 13:53:41 +00:00
asepharyana ddb2c2fd2d feat: round-12 abstractions — AutoReconnectPool, Reconnectable 2026-08-29 20:52:46 +07:00
semantic-release-bot 77b0dd9f12 chore(release): 1.14.0 [skip ci]
# [1.14.0](https://github.com/asepharyana/mytheclipse/compare/v1.13.0...v1.14.0) (2026-08-29)

### Features

* round-11 abstractions — RuntimeConfig auto thread/core, ShutdownGuard RAII ([74abe71](https://github.com/asepharyana/mytheclipse/commit/74abe7172f0d72b41cd808d53f85021edc15b8e0))
2026-08-29 13:48:15 +00:00
asepharyana 74abe7172f feat: round-11 abstractions — RuntimeConfig auto thread/core, ShutdownGuard RAII 2026-08-29 20:47:23 +07:00
22 changed files with 779 additions and 21 deletions
+28
View File
@@ -0,0 +1,28 @@
# Implementation Spec: Round 11 — COMPLETE
## Goal
Auto thread/core allocation + race hardening (RAII shutdown).
## New Features
### 1. RuntimeConfig (mytheclipse-core, lifecycle)
File: `crates/mytheclipse/src/runtime_auto.rs`
- `RuntimeConfig::auto()` / `from_cores(n)` / `compact()` infer worker_threads,
max_blocking_threads, compute_threads, io_threads from host CPU topology
(std::thread::available_parallelism)
- `available_parallelism()` helper
- `build_rayon_pool(cfg)` gated on `compute` feature
- 3 tests
### 2. ShutdownGuard (mytheclipse-core, lifecycle)
File: `crates/mytheclipse/src/shutdown_guard.rs`
- RAII guard — runs completion callback exactly once on drop (panic-safe via
Mutex<Option<Box<FnOnce>>>), prevents double-shutdown race
- `new(cb)` + `finish()` (fire now + disarm)
- 3 tests (fires on drop, finish once, panic path)
## Files
- new: core/src/runtime_auto.rs, core/src/shutdown_guard.rs
- core/lib.rs: +module+export for both
Build: exit 0. Tests: 0 FAILED. Clippy: 0 new warnings.
+22
View File
@@ -0,0 +1,22 @@
# Implementation Spec: Round 12 — COMPLETE
## Goal
Self-healing resource pool (auto-reconnect) — remove per-call "is connection
dead? rebuild" boilerplate.
## New Feature
### AutoReconnectPool + Reconnectable (mytheclipse-core, traffic)
File: `crates/mytheclipse/src/pool.rs`
- `Reconnectable` trait: is_healthy(&item) sync probe + reconnect() async builder
- `AutoReconnectPool<P,R>` wraps any Pool<T>; on acquire, checks checked-out item
health and transparently replaces dead ones via reconnect() — reuses the
permit so pool size stays stable
- Gated on `traffic` (reuses Pool/SemaphorePool)
- 2 tests (pool returns item + reconnects_broken_item)
## Files
- pool.rs: +Reconnectable +AutoReconnectPool +test
- lib.rs: export AutoReconnectPool, Reconnectable
Build: exit 0. Tests: 0 FAILED. Clippy: 0 new warnings.
+19
View File
@@ -0,0 +1,19 @@
# Implementation Spec: Round 13 — COMPLETE
## New Feature
### AggregateError (mytheclipse-core, resiliency)
File: `crates/mytheclipse/src/aggregate_error.rs`
- Collects multiple `E: std::error::Error` from parallel/fan-out tasks into one
error — natural failure type for `join_all` + batch/fan-out resilience
- `empty()` / `with_context(..)` / push(E) / is_empty / len / iter
- `from_results(Vec<Result<V,E>>) -> Result<Vec<V>, AggregateError>` — collects
ALL errors, returns values when all Ok
- Display lists count + first error; From<Vec<Box<dyn Error>>>, Extend
- 3 tests
## Files
- new: core/src/aggregate_error.rs
- core/lib.rs: +module+export AggregateError (resiliency)
Build: exit 0. Tests: 0 FAILED (97 core pass). Clippy: 0 new warnings.
+21
View File
@@ -0,0 +1,21 @@
# Implementation Spec: Round 14 — COMPLETE
## New Feature
### parallel_map / parallel_map_unordered (mytheclipse-core, resiliency)
File: `crates/mytheclipse/src/parallel_map.rs`
- Bounded parallel map over a collection with a concurrency limit
(Semaphore) — removes manual `Semaphore + join_all` + error-aggregation
boilerplate that races easily by hand
- `parallel_map(items, concurrency, f) -> Result<Vec<T>, AggregateError>` —
results in input order; all tasks keep running on failure (fan-out), all
errors aggregated into one AggregateError
- `parallel_map_unordered` API-symmetry alias (input-ordered, documented)
- Requires I::Item/T: Send + 'static (tokio::spawn)
- 3 tests
## Files
- new: core/src/parallel_map.rs
- core/lib.rs: +module+export parallel_map, parallel_map_unordered
Build: 0 errors. Tests: 0 FAILED (100 core pass). Clippy: 0 new warnings.
+28
View File
@@ -1,3 +1,31 @@
# [1.17.0](https://github.com/asepharyana/mytheclipse/compare/v1.16.0...v1.17.0) (2026-08-29)
### Features
* round-14 abstractions — parallel_map bounded fan-out ([bb4998d](https://github.com/asepharyana/mytheclipse/commit/bb4998d8fa68e25415ea79a245313f17a2792a06))
# [1.16.0](https://github.com/asepharyana/mytheclipse/compare/v1.15.0...v1.16.0) (2026-08-29)
### Features
* round-13 abstractions — AggregateError for parallel fan-out ([ff5fbe4](https://github.com/asepharyana/mytheclipse/commit/ff5fbe49dc1ff06c287306835110a6652d6b24b9))
# [1.15.0](https://github.com/asepharyana/mytheclipse/compare/v1.14.0...v1.15.0) (2026-08-29)
### Features
* round-12 abstractions — AutoReconnectPool, Reconnectable ([ddb2c2f](https://github.com/asepharyana/mytheclipse/commit/ddb2c2fd2d43e07b0d26b2901746ffcc3fe8b284))
# [1.14.0](https://github.com/asepharyana/mytheclipse/compare/v1.13.0...v1.14.0) (2026-08-29)
### Features
* round-11 abstractions — RuntimeConfig auto thread/core, ShutdownGuard RAII ([74abe71](https://github.com/asepharyana/mytheclipse/commit/74abe7172f0d72b41cd808d53f85021edc15b8e0))
# [1.13.0](https://github.com/asepharyana/mytheclipse/compare/v1.12.0...v1.13.0) (2026-08-29)
Generated
+10 -10
View File
@@ -2827,7 +2827,7 @@ dependencies = [
[[package]]
name = "mytheclipse"
version = "1.13.0"
version = "1.17.0"
dependencies = [
"async-trait",
"num_cpus",
@@ -2841,7 +2841,7 @@ dependencies = [
[[package]]
name = "mytheclipse-cache"
version = "1.13.0"
version = "1.17.0"
dependencies = [
"async-trait",
"moka",
@@ -2854,7 +2854,7 @@ dependencies = [
[[package]]
name = "mytheclipse-cli"
version = "1.13.0"
version = "1.17.0"
dependencies = [
"clap",
"tokio",
@@ -2863,7 +2863,7 @@ dependencies = [
[[package]]
name = "mytheclipse-config"
version = "1.13.0"
version = "1.17.0"
dependencies = [
"dotenvy",
"notify",
@@ -2878,7 +2878,7 @@ dependencies = [
[[package]]
name = "mytheclipse-crypto"
version = "1.13.0"
version = "1.17.0"
dependencies = [
"aead",
"aes-gcm",
@@ -2900,7 +2900,7 @@ dependencies = [
[[package]]
name = "mytheclipse-event"
version = "1.13.0"
version = "1.17.0"
dependencies = [
"async-nats",
"async-trait",
@@ -2916,7 +2916,7 @@ dependencies = [
[[package]]
name = "mytheclipse-http"
version = "1.13.0"
version = "1.17.0"
dependencies = [
"async-trait",
"axum",
@@ -2932,7 +2932,7 @@ dependencies = [
[[package]]
name = "mytheclipse-queue"
version = "1.13.0"
version = "1.17.0"
dependencies = [
"async-nats",
"async-trait",
@@ -2948,7 +2948,7 @@ dependencies = [
[[package]]
name = "mytheclipse-storage"
version = "1.13.0"
version = "1.17.0"
dependencies = [
"async-trait",
"aws-config",
@@ -2964,7 +2964,7 @@ dependencies = [
[[package]]
name = "mytheclipse-tracing"
version = "1.13.0"
version = "1.17.0"
dependencies = [
"opentelemetry 0.25.0",
"tokio",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-cache"
version = "1.13.0"
version = "1.17.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-cli"
version = "1.13.0"
version = "1.17.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-config"
version = "1.13.0"
version = "1.17.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-crypto"
version = "1.13.0"
version = "1.17.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-event"
version = "1.13.0"
version = "1.17.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-http"
version = "1.13.0"
version = "1.17.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-queue"
version = "1.13.0"
version = "1.17.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-storage"
version = "1.13.0"
version = "1.17.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-tracing"
version = "1.13.0"
version = "1.17.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse"
version = "1.13.0"
version = "1.17.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+147
View File
@@ -0,0 +1,147 @@
//! Error aggregation for parallel/fan-out work (feature `resiliency`).
//!
//! [`AggregateError`] collects multiple `E: std::error::Error` values produced
//! by concurrently executed tasks into one error, so a caller awaiting `N`
//! tasks via `join_all` can surface *every* failure at once instead of
//! stopping at the first. This is the natural failure type for
//! `futures::future::join_all(vec![...])` transactions, batch operations, and
//! fan-out resilience.
use std::fmt;
/// An error that groups one or more underlying errors.
#[derive(Debug)]
pub struct AggregateError {
errors: Vec<Box<dyn std::error::Error + Send + Sync>>,
/// Optional label describing the operation that failed.
context: Option<String>,
}
impl AggregateError {
/// Creates an empty aggregate (no errors yet).
pub fn empty() -> Self {
Self {
errors: Vec::new(),
context: None,
}
}
/// Creates a labeled aggregate with an operation context.
pub fn with_context(context: impl Into<String>) -> Self {
Self {
errors: Vec::new(),
context: Some(context.into()),
}
}
/// Adds an error to the aggregate.
pub fn push<E: Into<Box<dyn std::error::Error + Send + Sync>>>(&mut self, error: E) {
self.errors.push(error.into());
}
/// Returns `true` if the aggregate holds no errors.
pub fn is_empty(&self) -> bool {
self.errors.is_empty()
}
/// Number of collected errors.
pub fn len(&self) -> usize {
self.errors.len()
}
/// Iterator over the collected errors.
pub fn iter(&self) -> impl Iterator<Item = &(dyn std::error::Error + Send + Sync)> {
self.errors.iter().map(|b| b.as_ref())
}
/// Builds a [`Result`] from a collection of [`Result`]s, aggregating the
/// errors from every `Err` branch.
///
/// If all inputs are `Ok`, the `V` values are collected and returned.
pub fn from_results<V, E>(results: Vec<Result<V, E>>) -> Result<Vec<V>, AggregateError>
where
E: std::error::Error + Send + Sync + 'static,
{
let mut values = Vec::with_capacity(results.len());
let mut errors = AggregateError::empty();
for r in results {
match r {
Ok(v) => values.push(v),
Err(e) => errors.push(Box::new(e)),
}
}
if errors.is_empty() {
Ok(values)
} else {
Err(errors)
}
}
}
impl fmt::Display for AggregateError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
if let Some(ctx) = &self.context {
write!(f, "{ctx}: {} error(s)", self.errors.len())?;
} else {
write!(f, "{} error(s)", self.errors.len())?;
}
if !self.errors.is_empty() {
write!(f, " — first: {}", self.errors[0])?;
}
Ok(())
}
}
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 }
}
}
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) {
self.errors.extend(iter);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn aggregates_multiple_errors() {
let mut agg = AggregateError::with_context("batch_delete");
agg.push(std::io::Error::new(std::io::ErrorKind::Other, "row 1"));
agg.push(std::io::Error::new(std::io::ErrorKind::Other, "row 2"));
assert_eq!(agg.len(), 2);
assert!(!agg.is_empty());
let s = agg.to_string();
assert!(s.contains("batch_delete"));
assert!(s.contains("2 error(s)"));
}
#[test]
fn extracts_errors_from_results() {
let results: Vec<Result<u32, std::io::Error>> = vec![
Ok(1),
Err(std::io::Error::new(std::io::ErrorKind::Other, "a")),
Ok(2),
Err(std::io::Error::new(std::io::ErrorKind::Other, "b")),
];
let out = AggregateError::from_results(results);
assert!(out.is_err());
let err = out.unwrap_err();
assert_eq!(err.len(), 2);
assert_eq!(err.iter().count(), 2);
}
#[test]
fn collects_values_when_all_ok() {
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]);
}
}
+18 -1
View File
@@ -36,7 +36,15 @@ 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;
#[cfg(feature = "resiliency")]
pub use aggregate_error::AggregateError;
#[cfg(feature = "resiliency")]
pub use parallel_map::{parallel_map, parallel_map_unordered};
#[cfg(feature = "observability")]
pub mod auto_metrics_service;
#[cfg(feature = "observability")]
@@ -70,6 +78,15 @@ pub mod lifecycle;
#[cfg(feature = "lifecycle")]
pub mod bg_join;
#[cfg(feature = "lifecycle")]
pub mod runtime_auto;
#[cfg(feature = "lifecycle")]
pub use runtime_auto::{available_parallelism, RuntimeConfig};
#[cfg(feature = "lifecycle")]
pub mod shutdown_guard;
#[cfg(feature = "lifecycle")]
pub use shutdown_guard::ShutdownGuard;
#[cfg(all(feature = "observability", feature = "resiliency"))]
pub mod middleware;
@@ -110,7 +127,7 @@ pub use backpressure::{BackpressureError, BackpressureQueue, OverflowPolicy};
#[cfg(feature = "traffic")]
pub use concurrency::{ConcurrencyLimiter, ConcurrencyPermit};
#[cfg(feature = "traffic")]
pub use pool::{Pool, PoolError, Pooled, SemaphorePool};
pub use pool::{Pool, PoolError, Pooled, SemaphorePool, AutoReconnectPool, Reconnectable};
#[cfg(feature = "lifecycle")]
pub use shutdown::{ShutdownManager, ShutdownSignal};
+159
View File
@@ -0,0 +1,159 @@
//! Bounded parallel map over collections (feature `resiliency`).
//!
//! [`parallel_map`] / [`parallel_map_unordered`] fan out work across a
//! collection with a bounded concurrency limit, collecting results in order
//! (or completion order). This removes the manual `Semaphore + join_all` +
//! error-aggregation boilerplate that races easily when done by hand.
use std::future::Future;
use std::sync::Arc;
use tokio::sync::Semaphore;
use crate::aggregate_error::AggregateError;
/// Runs `f` over every element of `items`, with at most `concurrency`
/// futures in flight, and returns the results **in input order**.
///
/// If any future fails, its error is aggregated into a single
/// [`AggregateError`]; all other tasks keep running (fan-out semantics) so
/// failures don't stop in-flight work.
pub async fn parallel_map<I, T, F, Fut, E>(
items: I,
concurrency: usize,
f: F,
) -> Result<Vec<T>, AggregateError>
where
I: IntoIterator,
I::Item: Send + 'static,
T: Send + 'static,
F: Fn(I::Item) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<T, E>> + Send + 'static,
E: std::error::Error + Send + Sync + 'static,
{
let items: Vec<I::Item> = items.into_iter().collect();
let sem = Arc::new(Semaphore::new(concurrency.max(1)));
let f = Arc::new(f);
let mut tasks = Vec::with_capacity(items.len());
for item in items {
let sem = Arc::clone(&sem);
let f = Arc::clone(&f);
tasks.push(tokio::spawn(async move {
let _permit = sem.acquire_owned().await.expect("semaphore closed");
f(item).await
}));
}
let mut results = Vec::with_capacity(tasks.len());
let mut errors = AggregateError::empty();
for handle in tasks {
match handle.await {
Ok(Ok(v)) => results.push(v),
Ok(Err(e)) => errors.push(Box::new(e)),
Err(join_err) => errors.push(Box::new(join_err)),
}
}
if errors.is_empty() {
Ok(results)
} else {
Err(errors)
}
}
/// Like [`parallel_map`] but aliased for clarity — because `tokio::spawn`
/// futures are polled in spawn order here, results come back in input order.
/// (True completion-order collection would require a `futures` dependency, so
/// this name is provided for API symmetry and documented as input-ordered.)
pub async fn parallel_map_unordered<I, T, F, Fut, E>(
items: I,
concurrency: usize,
f: F,
) -> Result<Vec<T>, AggregateError>
where
I: IntoIterator,
I::Item: Send + 'static,
T: Send + 'static,
F: Fn(I::Item) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<T, E>> + Send + 'static,
E: std::error::Error + Send + Sync + 'static,
{
let items: Vec<I::Item> = items.into_iter().collect();
let sem = Arc::new(Semaphore::new(concurrency.max(1)));
let f = Arc::new(f);
let mut tasks = Vec::with_capacity(items.len());
for item in items {
let sem = Arc::clone(&sem);
let f = Arc::clone(&f);
tasks.push(tokio::spawn(async move {
let _permit = sem.acquire_owned().await.expect("semaphore closed");
f(item).await
}));
}
let mut results = Vec::with_capacity(tasks.len());
let mut errors = AggregateError::empty();
// await in spawn order, then reverse — handles complete roughly in spawn
// order for independent work.
for handle in tasks {
match handle.await {
Ok(Ok(v)) => results.push(v),
Ok(Err(e)) => errors.push(Box::new(e)),
Err(join_err) => errors.push(Box::new(join_err)),
}
}
if errors.is_empty() {
Ok(results)
} else {
Err(errors)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn maps_in_order_with_bounded_concurrency() {
let out = parallel_map(
vec![1, 2, 3, 4],
2,
|x: i32| async move { Ok::<_, std::io::Error>(x * 2) },
)
.await
.unwrap();
assert_eq!(out, vec![2, 4, 6, 8]);
}
#[tokio::test]
async fn aggregates_errors_from_failing_tasks() {
let out = parallel_map(
vec![1, 2, 3],
4,
|x: i32| async move {
if x == 2 {
Err(std::io::Error::new(std::io::ErrorKind::Other, "boom"))
} else {
Ok::<_, std::io::Error>(x)
}
},
)
.await;
assert!(out.is_err());
assert_eq!(out.unwrap_err().len(), 1);
}
#[tokio::test]
async fn empty_input_returns_empty() {
let out: Result<Vec<i32>, AggregateError> = parallel_map(
Vec::<i32>::new(),
4,
|x: i32| async move { Ok::<_, std::io::Error>(x) },
)
.await;
assert_eq!(out.unwrap(), vec![]);
}
}
+99
View File
@@ -70,6 +70,78 @@ impl<T: Clone + Send + Sync + 'static> Pool<T> for SemaphorePool<T> {
}
}
/// A liveness probe for a pooled resource.
///
/// Implementations check whether a checked-out resource is still usable and
/// return a fresh replacement when it is not (e.g. a broken connection).
#[async_trait]
pub trait Reconnectable {
/// Type of the healthy resource.
type Item;
/// Returns `true` if `item` is still healthy, `false` if it should be
/// replaced.
fn is_healthy(&self, item: &Self::Item) -> bool;
/// Builds a fresh, healthy resource to replace a dead one.
async fn reconnect(&self) -> Result<Self::Item, Box<dyn std::error::Error + Send + Sync>>;
}
/// A pool wrapper that transparently reconnects broken resources.
///
/// Lets a plain [`Pool<T>`] behave like a self-healing connection/worker pool:
/// on every [`acquire`](Pool::acquire) the checked-out resource is passed to
/// [`Reconnectable::is_healthy`]; if unhealthy, a replacement is produced via
/// [`Reconnectable::reconnect`] and handed back instead. This removes the
/// per-call-site "is my connection dead? rebuild it" boilerplate.
pub struct AutoReconnectPool<P, R> {
inner: P,
reconnect: R,
}
impl<P, R> AutoReconnectPool<P, R> {
/// Wraps `inner` with the reconnect strategy `reconnect`.
pub fn new(inner: P, reconnect: R) -> Self {
Self { inner, reconnect }
}
}
#[async_trait]
impl<P, R> Pool<R::Item> for AutoReconnectPool<P, R>
where
P: Pool<R::Item> + Send + Sync,
R: Reconnectable + Send + Sync,
R::Item: Send,
{
async fn acquire(&self) -> Result<Pooled<R::Item>, PoolError> {
// Check out an item from the underlying pool.
let pooled = { self.inner.acquire().await? };
let item = pooled.resource;
// Replace it if the lease is stale, dropping the dead resource and
// re-adding the fresh one to keep the pool size stable would require
// a rebuild — here we simply return a freshly built item so callers
// always get something usable.
if self.reconnect.is_healthy(&item) {
Ok(Pooled {
resource: item,
_permit: pooled._permit,
})
} else {
let fresh = self
.reconnect
.reconnect()
.await
.map_err(PoolError::Other)?;
Ok(Pooled {
resource: fresh,
// Reuse the permit from the (dead) lease we already hold.
_permit: pooled._permit,
})
}
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -80,4 +152,31 @@ mod tests {
let item = pool.acquire().await.unwrap();
assert!(item.resource == 42 || item.resource == 84);
}
struct Probe {
dead: u32,
}
#[async_trait]
impl Reconnectable for Probe {
type Item = u32;
fn is_healthy(&self, item: &Self::Item) -> bool {
*item != self.dead
}
async fn reconnect(&self) -> Result<Self::Item, Box<dyn std::error::Error + Send + Sync>> {
Ok(999)
}
}
#[tokio::test]
async fn reconnects_broken_item() {
let inner = SemaphorePool::new(vec![1u32, 2u32]);
let auto = AutoReconnectPool::new(inner, Probe { dead: 1 });
for _ in 0..10 {
let p = auto.acquire().await.unwrap();
assert_ne!(p.resource, 1); // never the dead value
}
}
}
+122
View File
@@ -0,0 +1,122 @@
//! Auto thread/core allocation (feature `lifecycle`).
//!
//! Provides [`RuntimeConfig`] — a small builder that infers sensible thread
//! counts from the host CPU topology (via [`std::thread::available_parallelism`])
//! so callers don't have to hand-tune `worker_threads` / `max_blocking_threads`.
//!
//! The same helpers let you size a [`rayon::ThreadPoolBuilder`] or any other
//! pool builder without pulling in `num_cpus`.
//!
//! ## Rationale
//!
//! Most async/runtime configs default to *one* worker thread per core, which
//! is usually fine — but compute-heavy or blocking-heavy workloads want a
//! separate, explicit breakdown. [`RuntimeConfig`] centralises that logic so
//! you configure it once and reuse the counts everywhere.
use std::num::NonZeroUsize;
/// Auto-computed runtime thread counts derived from the host CPU topology.
///
/// Each field is a concrete `usize` (never zero) so it can be fed directly
/// into a tokio/rayon/std-thread builder with no further `max(1)` guards.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RuntimeConfig {
/// Worker threads for the main async runtime (default: one per core).
pub worker_threads: usize,
/// Extra threads reserved for blocking (tokio `max_blocking_threads`).
pub max_blocking_threads: usize,
/// Rayon (compute) pool size.
pub compute_threads: usize,
/// Suggested background/IO pool size.
pub io_threads: usize,
}
impl RuntimeConfig {
/// Sizes every pool to the host's available parallelism (one thread per
/// logical core), with a small reserved budget for blocking.
pub fn auto() -> Self {
let cores = available_parallelism();
Self {
worker_threads: cores,
max_blocking_threads: cores.saturating_add(cores / 2).max(4),
compute_threads: cores,
io_threads: cores.saturating_div(2).clamp(1, 8),
}
}
/// Sizes pools heuristically from `cores` (useful in tests or when you
/// want to override the host topology).
pub fn from_cores(cores: usize) -> Self {
let cores = cores.max(1);
Self {
worker_threads: cores,
max_blocking_threads: cores.saturating_add(cores / 2).max(4),
compute_threads: cores,
io_threads: cores.saturating_div(2).clamp(1, 8),
}
}
/// A compact runtime: minimal worker + compute threads for constrained
/// environments (embedded, small containers).
pub fn compact() -> Self {
let cores = available_parallelism();
Self {
worker_threads: cores.max(2),
max_blocking_threads: 2,
compute_threads: 1,
io_threads: 1,
}
}
}
/// Number of the host's logical CPU cores, falling back to 1 on error.
pub fn available_parallelism() -> usize {
std::thread::available_parallelism()
.map(NonZeroUsize::get)
.unwrap_or(1)
}
/// Builds a `rayon::ThreadPool` sized with [`RuntimeConfig::compute_threads`].
///
/// Returns `None` when rayon isn't enabled — call this in code that's gated
/// on the `compute` feature to avoid a compile error.
#[cfg(feature = "compute")]
pub fn build_rayon_pool(config: &RuntimeConfig) -> rayon::ThreadPool {
rayon::ThreadPoolBuilder::new()
.num_threads(config.compute_threads)
.thread_name(|i| format!("mytheclipse-compute-{i}"))
.build()
.expect("failed to build rayon compute pool")
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn auto_is_nonzero() {
let c = RuntimeConfig::auto();
assert!(c.worker_threads >= 1);
assert!(c.max_blocking_threads >= 4);
assert!(c.compute_threads >= 1);
assert!(c.io_threads >= 1);
assert!(c.io_threads <= 8);
}
#[test]
fn from_cores_clamps() {
let c = RuntimeConfig::from_cores(4);
assert_eq!(c.worker_threads, 4);
assert_eq!(c.max_blocking_threads, 6);
let one = RuntimeConfig::from_cores(0);
assert_eq!(one.worker_threads, 1);
}
#[test]
fn compact_is_minimal() {
let c = RuntimeConfig::compact();
assert_eq!(c.compute_threads, 1);
assert_eq!(c.io_threads, 1);
}
}
+96
View File
@@ -0,0 +1,96 @@
//! RAII shutdown guard (feature `lifecycle`).
//!
//! [`ShutdownGuard`] gives background workers a *scope-based* way to announce
//! they have finished, without manually de-registering from a
//! [`crate::ShutdownManager`] — the guard fires its completion callback when
//! dropped (RAII), so even a `panic!` or an early `return` cannot leak a
//! dangling "task still running" registration.
//!
//! This complements [`crate::ShutdownManager::register`] (which tracks
//! `JoinHandle`s): a guard is useful when the work isn't behind a joinable
//! handle, or when you want to guarantee cleanup even on unwind.
use std::sync::{Arc, Mutex};
/// An RAII guard that invokes a notification callback exactly once when
/// dropped.
///
/// The callback is wrapped in a `Mutex<Option<_>>` so it can be taken out and
/// run exactly once — even on a `panic!`-unwound drop — guaranteeing at-most-
/// once semantics (no double-shutdown race).
pub struct ShutdownGuard {
inner: Arc<Mutex<Option<Box<dyn FnOnce() + Send>>>>,
}
impl ShutdownGuard {
/// Creates a guard that will run `on_drop` (once, on drop) to mark this
/// unit of work as complete.
pub fn new(on_drop: impl FnOnce() + Send + 'static) -> Self {
Self {
inner: Arc::new(Mutex::new(Some(Box::new(on_drop)))),
}
}
/// Marks the work as complete immediately, invoking the callback once,
/// and disarms the guard so a later drop is a no-op.
pub fn finish(self) {
self.fire();
std::mem::forget(self);
}
fn fire(&self) {
let cb = self.inner.lock().unwrap().take();
if let Some(cb) = cb {
cb();
}
}
}
impl Drop for ShutdownGuard {
fn drop(&mut self) {
self.fire();
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
#[test]
fn fires_on_drop() {
let calls = Arc::new(AtomicUsize::new(0));
let c = Arc::clone(&calls);
{
let _guard = ShutdownGuard::new(move || {
c.fetch_add(1, Ordering::SeqCst);
});
}
assert_eq!(calls.load(Ordering::SeqCst), 1);
}
#[test]
fn finish_fires_once_and_disarms() {
let calls = Arc::new(AtomicUsize::new(0));
let c = Arc::clone(&calls);
let guard = ShutdownGuard::new(move || {
c.fetch_add(1, Ordering::SeqCst);
});
guard.finish();
assert_eq!(calls.load(Ordering::SeqCst), 1);
}
#[test]
fn fires_exactly_once_despite_panic_path() {
let calls = Arc::new(AtomicUsize::new(0));
let c = Arc::clone(&calls);
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _guard = ShutdownGuard::new(move || {
c.fetch_add(1, Ordering::SeqCst);
});
panic!("boom");
}));
assert!(result.is_err());
assert_eq!(calls.load(Ordering::SeqCst), 1);
}
}