Compare commits

...
2 Commits
15 changed files with 197 additions and 20 deletions
+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.
+7
View File
@@ -1,3 +1,10 @@
# [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)
Generated
+10 -10
View File
@@ -2827,7 +2827,7 @@ dependencies = [
[[package]]
name = "mytheclipse"
version = "1.15.0"
version = "1.16.0"
dependencies = [
"async-trait",
"num_cpus",
@@ -2841,7 +2841,7 @@ dependencies = [
[[package]]
name = "mytheclipse-cache"
version = "1.15.0"
version = "1.16.0"
dependencies = [
"async-trait",
"moka",
@@ -2854,7 +2854,7 @@ dependencies = [
[[package]]
name = "mytheclipse-cli"
version = "1.15.0"
version = "1.16.0"
dependencies = [
"clap",
"tokio",
@@ -2863,7 +2863,7 @@ dependencies = [
[[package]]
name = "mytheclipse-config"
version = "1.15.0"
version = "1.16.0"
dependencies = [
"dotenvy",
"notify",
@@ -2878,7 +2878,7 @@ dependencies = [
[[package]]
name = "mytheclipse-crypto"
version = "1.15.0"
version = "1.16.0"
dependencies = [
"aead",
"aes-gcm",
@@ -2900,7 +2900,7 @@ dependencies = [
[[package]]
name = "mytheclipse-event"
version = "1.15.0"
version = "1.16.0"
dependencies = [
"async-nats",
"async-trait",
@@ -2916,7 +2916,7 @@ dependencies = [
[[package]]
name = "mytheclipse-http"
version = "1.15.0"
version = "1.16.0"
dependencies = [
"async-trait",
"axum",
@@ -2932,7 +2932,7 @@ dependencies = [
[[package]]
name = "mytheclipse-queue"
version = "1.15.0"
version = "1.16.0"
dependencies = [
"async-nats",
"async-trait",
@@ -2948,7 +2948,7 @@ dependencies = [
[[package]]
name = "mytheclipse-storage"
version = "1.15.0"
version = "1.16.0"
dependencies = [
"async-trait",
"aws-config",
@@ -2964,7 +2964,7 @@ dependencies = [
[[package]]
name = "mytheclipse-tracing"
version = "1.15.0"
version = "1.16.0"
dependencies = [
"opentelemetry 0.25.0",
"tokio",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-cache"
version = "1.15.0"
version = "1.16.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.15.0"
version = "1.16.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.15.0"
version = "1.16.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.15.0"
version = "1.16.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.15.0"
version = "1.16.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.15.0"
version = "1.16.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.15.0"
version = "1.16.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.15.0"
version = "1.16.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.15.0"
version = "1.16.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.15.0"
version = "1.16.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]);
}
}
+4
View File
@@ -36,7 +36,11 @@ pub mod retry;
#[cfg(feature = "resiliency")]
pub mod retry_ext;
#[cfg(feature = "resiliency")]
pub mod aggregate_error;
#[cfg(feature = "resiliency")]
pub use retry_ext::RetryExt;
#[cfg(feature = "resiliency")]
pub use aggregate_error::AggregateError;
#[cfg(feature = "observability")]
pub mod auto_metrics_service;
#[cfg(feature = "observability")]