Compare commits

...
17 Commits
Author SHA1 Message Date
semantic-release-bot ca39dea5db chore(release): 1.1.0 [skip ci]
# [1.1.0](https://github.com/asepharyana/mytheclipse/compare/v1.0.0...v1.1.0) (2026-08-28)

### Features

* add panic tracking and isolation with PanicTracker ([8dec413](https://github.com/asepharyana/mytheclipse/commit/8dec4130057420320500d7a9265ccb3cbfd0e030))
2026-08-28 11:45:57 +00:00
asepharyana 8dec413005 feat: add panic tracking and isolation with PanicTracker
- Implemented PanicTracker for logging panics using tracing.
- Introduced PanicInfo struct to capture panic details.
- Added tests for panic catching and hook restoration.

feat: implement token-bucket rate limiter

- Created RateLimiter for controlling token consumption with burst capacity.
- Added methods for acquiring tokens and checking available tokens.
- Included tests for rate limiting behavior and error handling.

feat: add automatic retry with exponential backoff

- Developed retry function to handle transient errors with configurable retry logic.
- Introduced RetryConfig for customizing retry parameters.
- Added tests for various retry scenarios and error handling.

feat: implement graceful shutdown coordination

- Created ShutdownManager to handle OS signals and manage task shutdown.
- Introduced ShutdownSignal for tasks to observe shutdown requests.
- Added tests for shutdown behavior and task management.

feat: add timeout functionality for async operations

- Implemented with_timeout function to enforce execution time limits.
- Created Timeout struct for wrapping futures with deadlines.
- Added tests for timeout behavior and error handling.
2026-08-28 18:45:20 +07:00
semantic-release-bot 4043a681db chore(release): 1.0.0 [skip ci]
# [1.0.0](https://github.com/asepharyana/mytheclipse/compare/v0.2.4...v1.0.0) (2026-08-28)

* feat!: rename CorexError to MytheclipseError ([7cdf525](https://github.com/asepharyana/mytheclipse/commit/7cdf52544c611b05c6365595a8ae713672706b67))

### BREAKING CHANGES

* CorexError is renamed to MytheclipseError.
2026-08-28 10:27:09 +00:00
asepharyana 7cdf52544c feat!: rename CorexError to MytheclipseError
Complete the mytheclipse rebranding: the public error type now uses the
mytheclipse name. Breaking API change (CorexError -> MytheclipseError),
hence the major version bump.

BREAKING CHANGE: CorexError is renamed to MytheclipseError.
2026-08-28 17:26:18 +07:00
asepharyana fa14529aa2 chore(repo): rename references to mytheclipse after GitHub repo rename
Update repository/homepage URLs in Cargo.toml and the gh api dispatch
repo path in .releaserc.json from asepharyana/corex to
asepharyana/mytheclipse (repo renamed on GitHub).
2026-08-28 17:22:23 +07:00
semantic-release-bot 4b6e9f35ef chore(release): 0.2.4 [skip ci]
## [0.2.4](https://github.com/asepharyana/corex/compare/v0.2.3...v0.2.4) (2026-08-28)

### Bug Fixes

* **release:** add actions: write permission for workflow dispatch ([5684938](https://github.com/asepharyana/corex/commit/5684938b4cd5a133a57ee835426035a804edcb38))
2026-08-28 10:19:23 +00:00
asepharyana 5684938b4c fix(release): add actions: write permission for workflow dispatch
gh api dispatch to publish.yml returned HTTP 403 because GITHUB_TOKEN
lacked actions: write. Add it to the release workflow permissions.
2026-08-28 17:18:45 +07:00
semantic-release-bot c830d2b949 chore(release): 0.2.3 [skip ci]
## [0.2.3](https://github.com/asepharyana/corex/compare/v0.2.2...v0.2.3) (2026-08-28)

### Bug Fixes

* **release:** hardcode repo in successCmd gh api ([d3c35ec](https://github.com/asepharyana/corex/commit/d3c35eccd0e3e1da21bcfed50ce2ba1f82d03cc5))
2026-08-28 10:17:43 +00:00
asepharyana d3c35eccd0 fix(release): hardcode repo in successCmd gh api
GITHUB_REPOSITORY is not in semantic-release template context, causing
ReferenceError in successCmd. Use the static repo path asepharyana/corex.
2026-08-28 17:17:17 +07:00
semantic-release-bot e7a932f716 chore(release): 0.2.2 [skip ci]
## [0.2.2](https://github.com/asepharyana/corex/compare/v0.2.1...v0.2.2) (2026-08-28)

### Bug Fixes

* **release:** dispatch publish via gh api in exec successCmd ([3c04aa4](https://github.com/asepharyana/corex/commit/3c04aa44b6110c8ce15278d7dc76828c44dfc8d9))
2026-08-28 10:15:48 +00:00
asepharyana 3c04aa44b6 fix(release): dispatch publish via gh api in exec successCmd
gh workflow run failed inside successCmd (exit 1, no output shown).
Switch to gh api REST dispatch which is transparent and reliable.
successCmd only runs after semantic-release actually publishes a new
version, so this fires once per release, exactly on the new tag.
Remove the separate Detect step (it depended on git state that is not
updated by semantic-release).
2026-08-28 17:15:12 +07:00
semantic-release-bot 5a2bb634a1 chore(release): 0.2.1 [skip ci]
## [0.2.1](https://github.com/asepharyana/corex/compare/v0.2.0...v0.2.1) (2026-08-28)

### Bug Fixes

* **release:** dispatch publish via exec successCmd not step output ([6c8c98e](https://github.com/asepharyana/corex/commit/6c8c98e86cf27b6a998124e3de3b226a78c82f66))
2026-08-28 10:10:44 +00:00
asepharyana 6c8c98e86c fix(release): dispatch publish via exec successCmd not step output
The prior pattern relying on steps.semantic-release.outputs did not
fire. Move the publish dispatch into @semantic-release/exec successCmd,
which runs only after a release is actually published and has
nextRelease.version available. Also add GH_TOKEN env for gh CLI
(GITHUB_TOKEN alone is not read by gh).
2026-08-28 17:10:16 +07:00
semantic-release-bot c995b2b937 chore(release): 0.2.0 [skip ci]
# [0.2.0](https://github.com/asepharyana/corex/compare/v0.1.0...v0.2.0) (2026-08-28)

### Bug Fixes

* **release:** remove --locked from cargo check in release prepare ([90322b5](https://github.com/asepharyana/corex/commit/90322b5f970aaf48b6da0adaf89c2f516aec0e4c))

### Features

* **release:** add semantic-release auto versioning ([12a1ab1](https://github.com/asepharyana/corex/commit/12a1ab10d04dec40f8b93edb51bb6104f6a75bcb))
2026-08-28 10:07:22 +00:00
asepharyana 90322b5f97 fix(release): remove --locked from cargo check in release prepare
cargo check --locked fails after bumping version in Cargo.toml because
Cargo.lock needs to be regenerated with the new root package version.
Semantic-release bump then committed Cargo.toml+CHANGELOG but lockfile
stayed stale, so the whole prepare step errored out.
2026-08-28 17:06:51 +07:00
asepharyana 8a3869773c chore(release): add workflow_dispatch trigger for manual release test 2026-08-28 17:05:28 +07:00
asepharyana 12a1ab10d0 feat(release): add semantic-release auto versioning
- Add .releaserc.json: conventional commits -> semantic version bump
  (major/minor/patch), updates Cargo.toml + Cargo.lock version, writes
  CHANGELOG.md, commits as 'chore(release): X.Y.Z [skip ci]'
- Add release.yml: runs semantic-release on main push, then triggers
  publish.yml via workflow_dispatch with the new tag
- Update publish.yml: add workflow_dispatch input (tag) so release
  workflow can trigger publish directly (GITHUB_TOKEN tag pushes do not
  trigger 'on: push: tags'); still supports manual tag pushes
2026-08-28 17:03:12 +07:00
22 changed files with 3252 additions and 53 deletions
+8
View File
@@ -50,6 +50,14 @@ jobs:
flags: "--no-default-features --features compute"
- name: bg only
flags: "--no-default-features --features bg"
- name: resiliency only
flags: "--no-default-features --features resiliency"
- name: traffic only
flags: "--no-default-features --features traffic"
- name: lifecycle only
flags: "--no-default-features --features lifecycle"
- name: observability only
flags: "--no-default-features --features observability"
steps:
- uses: actions/checkout@v4
- uses: dtolnay/rust-toolchain@stable
+14
View File
@@ -3,6 +3,15 @@ name: Publish to crates.io
on:
push:
tags: ["v*.*.*"]
workflow_dispatch:
inputs:
tag:
description: "Git tag to publish (e.g. v0.2.0). Defaults to latest tag."
required: false
type: string
permissions:
contents: read
jobs:
publish:
@@ -10,7 +19,12 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
with:
ref: ${{ inputs.tag || github.event.ref || github.ref }}
fetch-depth: 0
- uses: dtolnay/rust-toolchain@stable
- name: Publish mytheclipse
run: cargo publish --allow-dirty --no-verify
env:
+46
View File
@@ -0,0 +1,46 @@
name: Release
on:
push:
branches: [main]
workflow_dispatch:
permissions:
contents: write
issues: write
pull-requests: write
actions: write
jobs:
release:
name: Semantic Release
runs-on: ubuntu-latest
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Setup Rust toolchain
uses: dtolnay/rust-toolchain@stable
- name: Setup Node.js
uses: actions/setup-node@v4
with:
node-version: "lts/*"
- name: Install semantic-release
run: |
npm install -D \
semantic-release \
@semantic-release/exec \
@semantic-release/git \
@semantic-release/changelog \
@semantic-release/github
- name: Run semantic-release
id: semantic-release
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: npx semantic-release
+34
View File
@@ -0,0 +1,34 @@
{
"branches": [
"main"
],
"plugins": [
"@semantic-release/commit-analyzer",
"@semantic-release/release-notes-generator",
"@semantic-release/changelog",
[
"@semantic-release/exec",
{
"prepareCmd": "sed -i 's/^version = \\\"[^\\\"]*\\\"/version = \\\"${nextRelease.version}\\\"/' Cargo.toml && cargo check",
"successCmd": "gh api -X POST repos/asepharyana/mytheclipse/actions/workflows/publish.yml/dispatches -f ref=main -f 'inputs[tag]=v${nextRelease.version}'"
}
],
[
"@semantic-release/git",
{
"assets": [
"Cargo.toml",
"Cargo.lock",
"CHANGELOG.md"
],
"message": "chore(release): ${nextRelease.version} [skip ci]\n\n${nextRelease.notes}"
}
],
[
"@semantic-release/github",
{
"assets": []
}
]
]
}
+56
View File
@@ -0,0 +1,56 @@
# [1.1.0](https://github.com/asepharyana/mytheclipse/compare/v1.0.0...v1.1.0) (2026-08-28)
### Features
* add panic tracking and isolation with PanicTracker ([8dec413](https://github.com/asepharyana/mytheclipse/commit/8dec4130057420320500d7a9265ccb3cbfd0e030))
# [1.0.0](https://github.com/asepharyana/mytheclipse/compare/v0.2.4...v1.0.0) (2026-08-28)
* feat!: rename CorexError to MytheclipseError ([7cdf525](https://github.com/asepharyana/mytheclipse/commit/7cdf52544c611b05c6365595a8ae713672706b67))
### BREAKING CHANGES
* CorexError is renamed to MytheclipseError.
## [0.2.4](https://github.com/asepharyana/corex/compare/v0.2.3...v0.2.4) (2026-08-28)
### Bug Fixes
* **release:** add actions: write permission for workflow dispatch ([5684938](https://github.com/asepharyana/corex/commit/5684938b4cd5a133a57ee835426035a804edcb38))
## [0.2.3](https://github.com/asepharyana/corex/compare/v0.2.2...v0.2.3) (2026-08-28)
### Bug Fixes
* **release:** hardcode repo in successCmd gh api ([d3c35ec](https://github.com/asepharyana/corex/commit/d3c35eccd0e3e1da21bcfed50ce2ba1f82d03cc5))
## [0.2.2](https://github.com/asepharyana/corex/compare/v0.2.1...v0.2.2) (2026-08-28)
### Bug Fixes
* **release:** dispatch publish via gh api in exec successCmd ([3c04aa4](https://github.com/asepharyana/corex/commit/3c04aa44b6110c8ce15278d7dc76828c44dfc8d9))
## [0.2.1](https://github.com/asepharyana/corex/compare/v0.2.0...v0.2.1) (2026-08-28)
### Bug Fixes
* **release:** dispatch publish via exec successCmd not step output ([6c8c98e](https://github.com/asepharyana/corex/commit/6c8c98e86cf27b6a998124e3de3b226a78c82f66))
# [0.2.0](https://github.com/asepharyana/corex/compare/v0.1.0...v0.2.0) (2026-08-28)
### Bug Fixes
* **release:** remove --locked from cargo check in release prepare ([90322b5](https://github.com/asepharyana/corex/commit/90322b5f970aaf48b6da0adaf89c2f516aec0e4c))
### Features
* **release:** add semantic-release auto versioning ([12a1ab1](https://github.com/asepharyana/corex/commit/12a1ab10d04dec40f8b93edb51bb6104f6a75bcb))
Generated
+72 -1
View File
@@ -61,6 +61,17 @@ dependencies = [
"windows-sys",
]
[[package]]
name = "getrandom"
version = "0.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0"
dependencies = [
"cfg-if",
"libc",
"wasi",
]
[[package]]
name = "hermit-abi"
version = "0.5.2"
@@ -107,9 +118,10 @@ dependencies = [
[[package]]
name = "mytheclipse"
version = "0.1.0"
version = "1.1.0"
dependencies = [
"num_cpus",
"rand",
"rayon",
"tokio",
"tracing",
@@ -170,6 +182,15 @@ version = "0.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
[[package]]
name = "ppv-lite86"
version = "0.2.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9"
dependencies = [
"zerocopy",
]
[[package]]
name = "proc-macro2"
version = "1.0.107"
@@ -188,6 +209,36 @@ dependencies = [
"proc-macro2",
]
[[package]]
name = "rand"
version = "0.8.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e058c7de0b26af77780c769414d6257830bb240f3c38477dbc2c16e5f54d6d4c"
dependencies = [
"libc",
"rand_chacha",
"rand_core",
]
[[package]]
name = "rand_chacha"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88"
dependencies = [
"ppv-lite86",
"rand_core",
]
[[package]]
name = "rand_core"
version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c"
dependencies = [
"getrandom",
]
[[package]]
name = "rayon"
version = "1.12.0"
@@ -406,3 +457,23 @@ checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
dependencies = [
"windows-link",
]
[[package]]
name = "zerocopy"
version = "0.8.56"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb"
dependencies = [
"zerocopy-derive",
]
[[package]]
name = "zerocopy-derive"
version = "0.8.56"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.119",
]
+11 -6
View File
@@ -1,21 +1,22 @@
[package]
name = "mytheclipse"
version = "0.1.0"
version = "1.1.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
repository = "https://github.com/asepharyana/corex"
homepage = "https://github.com/asepharyana/corex"
repository = "https://github.com/asepharyana/mytheclipse"
homepage = "https://github.com/asepharyana/mytheclipse"
documentation = "https://docs.rs/mytheclipse"
authors = ["asepharyana <superaseph@gmail.com>"]
description = "Resource-aware abstractions for async I/O, heavy compute, and background queue management."
description = "Resource-aware abstractions for async I/O, heavy compute, background queue management, resiliency, traffic control, lifecycle, and observability."
readme = "README.md"
keywords = ["async", "concurrency", "rayon", "tokio", "resource-management"]
keywords = ["async", "concurrency", "rayon", "tokio", "resource-management", "resiliency", "retry", "circuit-breaker", "rate-limit", "observability", "cron", "shutdown"]
categories = ["asynchronous", "concurrency", "rust-patterns"]
[dependencies]
tokio = { version = "1.53", features = ["full"], optional = true }
rayon = { version = "1.12", optional = true }
rand = { version = "0.8", optional = true }
num_cpus = "1.17"
tracing = "0.1"
@@ -28,7 +29,11 @@ default = []
io = ["dep:tokio"]
compute = ["dep:rayon"]
bg = ["dep:tokio"]
full = ["io", "compute", "bg"]
resiliency = ["dep:tokio", "dep:rand"]
traffic = ["dep:tokio"]
lifecycle = ["dep:tokio"]
observability = ["dep:tokio"]
full = ["io", "compute", "bg", "resiliency", "traffic", "lifecycle", "observability"]
[[example]]
name = "main"
+102 -20
View File
@@ -4,7 +4,7 @@
[![Documentation](https://docs.rs/mytheclipse/badge.svg)](https://docs.rs/mytheclipse)
[![License](https://img.shields.io/badge/license-MIT%20OR%20Apache--2.0-blue.svg)](LICENSE-MIT)
Resource-aware execution primitives for Rust: async I/O, heavy compute, and background queue management, sized automatically from the host's logical core count and exposed through a single, lazily-initialized engine context.
Resource-aware execution primitives and reliability abstractions for Rust: async I/O, heavy compute, background queue management, resiliency, traffic control, lifecycle management, and observability. Sized automatically from the host's logical core count and exposed through a single, lazily-initialized engine context alongside self-contained, constructible utilities.
## Resource Sizing
@@ -16,12 +16,28 @@ Given $N$ logical cores (via `num_cpus::get()`):
| **Compute** | $\max(1, N - 1)$ | 7 | Sized `rayon::ThreadPool` + `catch_unwind` |
| **Background Queue** | $\max(2, \lfloor N / 2 \rfloor)$ | 4 | `tokio::sync::Semaphore` + `tokio::spawn` |
The three execution primitives (`io`, `compute`, `bg`) are sized from the host's CPU core count through `mytheclipse::context()`. Resiliency, traffic control, lifecycle, and observability utilities are constructible per-instance without global state.
## Features
- **`io`**: enables `mytheclipse::spawn_io`, instrumented async task spawning.
- **`compute`**: enables `mytheclipse::compute`, panic-isolated execution on a sized Rayon pool.
- **`bg`**: enables `mytheclipse::spawn_bg`, semaphore-bounded background tasks.
- **`full`**: enables all three subsystems.
- **`resiliency`**: fault tolerance abstractions:
- `retry` — auto-retry with exponential backoff and full/equal jitter.
- `CircuitBreaker` — failure threshold and cooldown state machine (Closed/Open/HalfOpen).
- `with_timeout` / `timeout` / `Timeout` — hard execution deadlines.
- **`traffic`**: traffic & load control:
- `RateLimiter` — lazy token-bucket rate limiter with burst capacity.
- `BackpressureQueue` — bounded queue with `DropOldest`, `Reject`, and `Block` overflow policies.
- `ConcurrencyLimiter` — synchronous RAII semaphore wrapper for capping concurrent operations.
- **`lifecycle`**: system lifecycle coordination:
- `ShutdownManager` / `ShutdownSignal` — OS signal catching (SIGINT/SIGTERM/Ctrl-C) and graceful task draining.
- `CronSchedule` / `schedule` — self-contained 5-field cron parser and async timer scheduler.
- **`observability`**: runtime visibility:
- `MetricsCollector` — thread-safe statistics collector with Prometheus text exposition format export.
- `PanicTracker` — non-fatal panic logging with tracing context and boundary isolation.
- **`full`**: enables all subsystems: `io`, `compute`, `bg`, `resiliency`, `traffic`, `lifecycle`, `observability`.
Zero features enabled by default (`default = []`), so you only pull in the dependencies your application actually uses.
@@ -31,40 +47,106 @@ Add to your `Cargo.toml`:
```toml
[dependencies]
mytheclipse = { version = "0.1", features = ["full"] }
mytheclipse = { version = "0.2", features = ["full"] }
```
Use the entry points directly:
### 1. Execution Primitives
```rust
#[tokio::main]
async fn main() {
// Optional explicit bootstrap: logs or validates resource sizing upfront.
// Omit it and the first call to any primitive below will initialize it lazily.
let ctx = mytheclipse::init();
println!(
"io_threads={} compute_threads={} bg_concurrency={}",
ctx.io_threads, ctx.compute_threads, ctx.bg_concurrency
);
// 1. Async I/O (instrumented with tracing)
let io = mytheclipse::spawn_io(async {
// ... network / disk work ...
42
});
// Async I/O (instrumented with tracing)
let io = mytheclipse::spawn_io(async { 42 });
// 2. Heavy Compute (isolated from worker panics)
// Heavy Compute (isolated from worker panics)
let sum = mytheclipse::compute(|| (1..=1_000_000u64).sum::<u64>())?;
// 3. Background Queue (concurrency-bounded)
let bg = mytheclipse::spawn_bg(async {
// ... deferred cleanup / telemetry ...
}).await;
// Background Queue (concurrency-bounded)
let bg = mytheclipse::spawn_bg(async { /* task */ }).await;
let _ = (io.await, bg.await);
}
```
### 2. Resiliency & Fault Tolerance
```rust
use std::time::Duration;
// Auto-Retry with Exponential Backoff + Jitter
let result = mytheclipse::retry(
mytheclipse::RetryConfig::default(),
|err| err.is_transient(),
|| async { make_network_request().await },
).await?;
// Circuit Breaker
let breaker = mytheclipse::CircuitBreaker::new(mytheclipse::CircuitBreakerConfig::default());
let value = breaker.call(|| fetch_remote_resource())?;
// Timeout & Deadlines
let value = mytheclipse::with_timeout(Duration::from_secs(5), async {
long_running_task().await
}).await?;
```
### 3. Traffic & Resource Control
```rust
// Rate Limiter (Token Bucket)
let limiter = mytheclipse::RateLimiter::new(100.0, 10);
limiter.acquire().await?;
// Concurrency Limiter (Sync Semaphore)
let limiter = mytheclipse::ConcurrencyLimiter::new(10);
let _permit = limiter.acquire(); // released on drop
// Backpressure Queue (Graceful Degradation)
let queue = mytheclipse::BackpressureQueue::new(100, mytheclipse::OverflowPolicy::DropOldest);
queue.push(job).await?;
let next_job = queue.pop().await;
```
### 4. Lifecycle & State Management
```rust
// Graceful Shutdown Manager
let shutdown = mytheclipse::ShutdownManager::new();
let sig = shutdown.handle();
tokio::spawn(async move {
let mut sig = sig;
tokio::select! {
_ = sig.wait() => { /* clean up */ }
_ = worker_loop() => {}
}
});
shutdown.drain(Duration::from_secs(10)).await;
// Cron Periodic Job Scheduler (Self-contained, no external crates)
let cron = mytheclipse::CronSchedule::parse("0 1 * * *")?; // 1 AM daily
let job = mytheclipse::schedule("0 1 * * *", || async {
clean_cache().await;
})?;
```
### 5. Telemetry & Observability
```rust
// Centralized Metrics Collector (Prometheus text exposition format)
let metrics = mytheclipse::MetricsCollector::new();
metrics.record_task(Duration::from_millis(15));
metrics.inc_counter("http_requests_total", 1);
let prometheus_output = metrics.export_prometheus();
// Panic Isolation Tracker
let guard = mytheclipse::PanicTracker::install(); // logs panics with tracing span
let result = mytheclipse::PanicTracker::catch(|| {
risky_operation()
});
```
## Running the Example
```bash
+150 -6
View File
@@ -1,10 +1,14 @@
//! Demonstrates `mytheclipse`'s three execution primitives end to end, including
//! automatic recovery from a panicking compute closure.
//! Demonstrates `mytheclipse`'s capabilities end to end:
//! execution primitives, resiliency & fault tolerance, traffic control,
//! lifecycle management, and observability.
use std::time::Duration;
#[tokio::main]
async fn main() {
tracing_subscriber::fmt::init();
println!("=== 1. Execution Primitives & Context ===");
let ctx = mytheclipse::init();
println!(
"engine context: io_threads={} compute_threads={} bg_concurrency={}",
@@ -12,20 +16,20 @@ async fn main() {
);
let io_handle = mytheclipse::spawn_io(async {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
tokio::time::sleep(Duration::from_millis(10)).await;
42u64
});
let sum_result = mytheclipse::compute(|| (1..=1_000u64).sum::<u64>());
let panic_result: Result<u64, mytheclipse::CorexError> = mytheclipse::compute(|| {
panic!("intentional panic to demonstrate isolation");
let panic_result: Result<u64, mytheclipse::MytheclipseError> = mytheclipse::compute(|| {
panic!("intentional panic to demonstrate compute isolation");
});
let recovery_result = mytheclipse::compute(|| 2u64 + 2u64);
let bg_handle = mytheclipse::spawn_bg(async {
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
tokio::time::sleep(Duration::from_millis(10)).await;
"bg-task-done"
})
.await;
@@ -38,4 +42,144 @@ async fn main() {
println!("compute panic-isolation result: {panic_result:?}");
println!("compute pool still usable after panic: {recovery_result:?}");
println!("spawn_bg result: {bg_value}");
println!("\n=== 2. Resiliency & Fault Tolerance ===");
// Auto-Retry with Exponential Backoff & Full Jitter
let attempts = std::sync::atomic::AtomicU32::new(0);
let retry_res = mytheclipse::retry(
mytheclipse::RetryConfig {
max_attempts: 4,
base_delay: Duration::from_millis(10),
..Default::default()
},
|_| true,
|| async {
let count = attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
if count < 3 {
Err("temporary network failure")
} else {
Ok("connected successfully!")
}
},
)
.await;
println!(
"retry result (succeeded on attempt {}): {retry_res:?}",
attempts.load(std::sync::atomic::Ordering::SeqCst)
);
// Circuit Breaker
let breaker = mytheclipse::CircuitBreaker::new(mytheclipse::CircuitBreakerConfig {
failure_threshold: 2,
open_timeout: Duration::from_millis(50),
..Default::default()
});
let _ = breaker.call(|| Err::<(), _>("service unavailable"));
let _ = breaker.call(|| Err::<(), _>("service unavailable"));
println!(
"circuit breaker state after 2 failures: {:?}",
breaker.state()
);
let call_when_open: Result<(), mytheclipse::CircuitError<&str>> =
breaker.call(|| Ok::<(), &str>(()));
println!("call blocked by circuit breaker: {call_when_open:?}");
// Timeout & Deadlines
let timeout_res = mytheclipse::with_timeout(Duration::from_millis(20), async {
tokio::time::sleep(Duration::from_millis(100)).await;
"finished"
})
.await;
println!("with_timeout result on slow task: {timeout_res:?}");
println!("\n=== 3. Traffic & Resource Control ===");
// Rate Limiter (Token Bucket)
let limiter = mytheclipse::RateLimiter::new(100.0, 2);
println!(
"rate limiter initial tokens: {}",
limiter.available_tokens()
);
let _ = limiter.try_acquire();
println!(
"rate limiter tokens after 1 acquire: {}",
limiter.available_tokens()
);
// Concurrency Limiter (Sync Semaphore Wrapper)
let conc = mytheclipse::ConcurrencyLimiter::new(2);
let permit1 = conc.try_acquire().expect("permit 1");
let permit2 = conc.try_acquire().expect("permit 2");
println!(
"concurrency limiter in use: {}/{}",
conc.in_use(),
conc.max()
);
assert!(conc.try_acquire().is_none());
drop(permit1);
drop(permit2);
println!(
"concurrency limiter in use after drop: {}/{}",
conc.in_use(),
conc.max()
);
// Backpressure Queue (Graceful Degradation)
let queue = mytheclipse::BackpressureQueue::new(2, mytheclipse::OverflowPolicy::DropOldest);
queue.push("item-1").await.unwrap();
queue.push("item-2").await.unwrap();
queue.push("item-3").await.unwrap(); // item-1 will be dropped
println!(
"backpressure queue: len={}, dropped={}, next={:?}",
queue.len(),
queue.dropped_count(),
queue.pop().await
);
println!("\n=== 4. Lifecycle & State Management ===");
// Cron Scheduler (Minimal 5-field parser without external crate)
let cron = mytheclipse::CronSchedule::parse("0 1 * * *").expect("valid cron expression");
let now = mytheclipse::cron::CronTime::now();
let next_runs = cron.next_five(now);
println!("cron '0 1 * * *' next 3 scheduled fire times:");
for t in next_runs.iter().take(3) {
println!(
" -> {:04}-{:02}-{:02} {:02}:{:02}:00 UTC",
t.year, t.month, t.day, t.hour, t.minute
);
}
// Graceful Shutdown Manager
let shutdown = mytheclipse::ShutdownManager::new();
let sig = shutdown.handle();
println!(
"shutdown signal status before request: is_shutdown={}",
sig.is_shutdown()
);
shutdown.request();
println!(
"shutdown signal status after request: is_shutdown={}",
sig.is_shutdown()
);
println!("\n=== 5. Telemetry & Observability ===");
// Centralized Metrics Collector (Prometheus text exposition format)
let metrics = mytheclipse::MetricsCollector::new();
metrics.record_task(Duration::from_millis(15));
metrics.record_task(Duration::from_millis(25));
metrics.set_active_threads(ctx.compute_threads);
metrics.set_queue_capacity(100);
metrics.set_queue_remaining(98);
metrics.inc_counter("http_requests_total", 42);
metrics.set_gauge("memory_usage_mb", 128.5);
println!("Prometheus Export Output:\n---");
print!("{}", metrics.export_prometheus());
println!("---");
// Panic & Span Isolation Tracker
let isolated_panic = mytheclipse::PanicTracker::catch(|| {
panic!("worker routine isolated error");
});
println!("PanicTracker::catch result: {isolated_panic:?}");
println!("\nAll mytheclipse features executed successfully!");
}
+310
View File
@@ -0,0 +1,310 @@
//! A bounded queue with configurable overflow handling (feature `traffic`).
//!
//! [`BackpressureQueue`] buffers items up to a fixed capacity. When full, the
//! configured [`OverflowPolicy`] decides the graceful-degradation behavior:
//! drop the oldest item to make room, reject the new item and hand it back to
//! the caller, or block the caller until a slot frees up.
use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use tokio::sync::Notify;
use tracing::Instrument;
/// Behaviour applied when the queue is at capacity.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OverflowPolicy {
/// Discard the oldest item to free a slot, then accept the new one.
DropOldest,
/// Return the new item to the caller as an error without enqueueing it.
Reject,
/// Block the calling coroutine until a slot frees up.
Block,
}
/// The error returned when an item cannot be enqueued.
#[derive(Debug)]
pub enum BackpressureError<T> {
/// The queue was full and the [`OverflowPolicy::Reject`] policy declined
/// the item. The rejected item is returned so the caller may handle it.
QueueFull(T),
}
impl<T> std::fmt::Display for BackpressureError<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "backpressure queue is full")
}
}
impl<T: std::fmt::Debug> std::error::Error for BackpressureError<T> {}
struct Inner<T> {
queue: Mutex<VecDeque<T>>,
not_full: Notify,
not_empty: Notify,
capacity: usize,
policy: OverflowPolicy,
accepted: AtomicU64,
dropped: AtomicU64,
rejected: AtomicU64,
closed: AtomicU64,
}
const CLOSED: u64 = 1;
const OPEN: u64 = 0;
/// A thread-safe, bounded work queue with graceful-degradation overflow
/// handling.
///
/// Construct with [`BackpressureQueue::new`]; await [`push`](Self::push) /
/// [`pop`](Self::pop) in async contexts, or use
/// [`try_push`](Self::try_push) to fail fast.
#[derive(Clone)]
pub struct BackpressureQueue<T> {
inner: Arc<Inner<T>>,
}
impl<T> BackpressureQueue<T> {
/// Builds a queue with `capacity` slots and the given overflow policy.
///
/// # Panics
///
/// Panics if `capacity` is zero.
pub fn new(capacity: usize, policy: OverflowPolicy) -> Self {
assert!(capacity > 0, "backpressure queue capacity must be > 0");
Self {
inner: Arc::new(Inner {
queue: Mutex::new(VecDeque::with_capacity(capacity)),
not_full: Notify::new(),
not_empty: Notify::new(),
capacity,
policy,
accepted: AtomicU64::new(0),
dropped: AtomicU64::new(0),
rejected: AtomicU64::new(0),
closed: AtomicU64::new(OPEN),
}),
}
}
/// Enqueues `item`, applying the configured overflow policy when full.
///
/// For [`OverflowPolicy::Block`], awaits a free slot first (in a
/// `mytheclipse_backpressure_task` span); for [`OverflowPolicy::Reject`],
/// returns the item back as [`BackpressureError::QueueFull`] without
/// blocking; for [`OverflowPolicy::DropOldest`], never fails and evicts
/// the oldest item.
pub async fn push(&self, item: T) -> Result<(), BackpressureError<T>> {
if self.inner.closed.load(Ordering::Acquire) == CLOSED {
return Err(BackpressureError::QueueFull(item));
}
match self.inner.policy {
OverflowPolicy::Block => {
let span = tracing::info_span!("mytheclipse_backpressure_task");
loop {
{
let mut queue = self.inner.queue.lock().unwrap();
if queue.len() < self.inner.capacity {
queue.push_back(item);
self.inner.accepted.fetch_add(1, Ordering::AcqRel);
drop(queue);
self.inner.not_empty.notify_one();
return Ok(());
}
}
let waiter = self.inner.not_full.notified();
tokio::pin!(waiter);
waiter.as_mut().instrument(span.clone()).await;
}
}
OverflowPolicy::Reject => self.try_push(item),
OverflowPolicy::DropOldest => {
self.try_push_dropping(item);
Ok(())
}
}
}
/// Non-blocking enqueue honoring [`OverflowPolicy::DropOldest`] and
/// [`OverflowPolicy::Reject`]; for [`OverflowPolicy::Block`] it behaves
/// like [`OverflowPolicy::Reject`] (never blocks).
pub fn try_push(&self, item: T) -> Result<(), BackpressureError<T>> {
let mut queue = self.inner.queue.lock().unwrap();
if self.inner.closed.load(Ordering::Acquire) == CLOSED {
return Err(BackpressureError::QueueFull(item));
}
if queue.len() >= self.inner.capacity {
match self.inner.policy {
OverflowPolicy::DropOldest => {
let _dropped = queue.pop_front();
self.inner.dropped.fetch_add(1, Ordering::AcqRel);
}
_ => {
self.inner.rejected.fetch_add(1, Ordering::AcqRel);
return Err(BackpressureError::QueueFull(item));
}
}
}
queue.push_back(item);
self.inner.accepted.fetch_add(1, Ordering::AcqRel);
drop(queue);
self.inner.not_empty.notify_one();
Ok(())
}
fn try_push_dropping(&self, item: T) {
let mut queue = self.inner.queue.lock().unwrap();
if queue.len() >= self.inner.capacity {
let _dropped = queue.pop_front();
self.inner.dropped.fetch_add(1, Ordering::AcqRel);
}
queue.push_back(item);
self.inner.accepted.fetch_add(1, Ordering::AcqRel);
drop(queue);
self.inner.not_empty.notify_one();
}
/// Awaits the next item, blocking until one is available.
pub async fn pop(&self) -> Option<T> {
loop {
{
let mut queue = self.inner.queue.lock().unwrap();
if let Some(item) = queue.pop_front() {
self.inner.not_full.notify_one();
return Some(item);
}
if self.inner.closed.load(Ordering::Acquire) == CLOSED && queue.is_empty() {
return None;
}
}
self.inner.not_empty.notified().await;
}
}
/// Attempts to pop an item without blocking.
pub fn try_pop(&self) -> Option<T> {
let mut queue = self.inner.queue.lock().unwrap();
let item = queue.pop_front();
if item.is_some() {
self.inner.not_full.notify_one();
}
item
}
/// The number of items currently buffered.
pub fn len(&self) -> usize {
self.inner.queue.lock().unwrap().len()
}
/// Whether the queue is empty.
pub fn is_empty(&self) -> bool {
self.len() == 0
}
/// The queue's capacity.
pub fn capacity(&self) -> usize {
self.inner.capacity
}
/// How much room remains.
pub fn remaining(&self) -> usize {
self.inner.capacity - self.len()
}
/// Number of items successfully enqueued.
pub fn accepted_count(&self) -> u64 {
self.inner.accepted.load(Ordering::Acquire)
}
/// Number of items evicted by drop policies.
pub fn dropped_count(&self) -> u64 {
self.inner.dropped.load(Ordering::Acquire)
}
/// Number of items rejected by the reject/closed paths.
pub fn rejected_count(&self) -> u64 {
self.inner.rejected.load(Ordering::Acquire)
}
/// Closes the queue: no further items are accepted and `pop` drains the
/// remaining items then returns `None`.
pub fn close(&self) {
self.inner.closed.store(CLOSED, Ordering::Release);
self.inner.not_empty.notify_waiters();
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
#[tokio::test]
async fn push_pop_roundtrip() {
let q: BackpressureQueue<u32> = BackpressureQueue::new(4, OverflowPolicy::Reject);
q.push(1).await.unwrap();
q.push(2).await.unwrap();
assert_eq!(q.pop().await, Some(1));
assert_eq!(q.pop().await, Some(2));
assert!(q.is_empty());
}
#[tokio::test]
async fn reject_returns_item_when_full() {
let q: BackpressureQueue<u32> = BackpressureQueue::new(2, OverflowPolicy::Reject);
q.push(1).await.unwrap();
q.push(2).await.unwrap();
let err = q.push(3).await.unwrap_err();
match err {
BackpressureError::QueueFull(v) => assert_eq!(v, 3),
}
assert_eq!(q.len(), 2);
assert_eq!(q.rejected_count(), 1);
}
#[tokio::test]
async fn drop_oldest_evicts_front() {
let q: BackpressureQueue<u32> = BackpressureQueue::new(2, OverflowPolicy::DropOldest);
q.push(1).await.unwrap();
q.push(2).await.unwrap();
q.push(3).await.unwrap();
assert_eq!(q.len(), 2);
assert_eq!(q.pop().await, Some(2)); // 1 was dropped
assert_eq!(q.dropped_count(), 1);
}
#[tokio::test]
async fn block_waits_for_a_slot() {
let q: BackpressureQueue<u32> = BackpressureQueue::new(1, OverflowPolicy::Block);
q.push(1).await.unwrap();
// A second push must block until pop frees a slot.
let q2 = q.clone();
let pusher = tokio::spawn(async move {
q2.push(2).await.unwrap();
});
tokio::time::sleep(Duration::from_millis(20)).await;
assert_eq!(q.pop().await, Some(1));
pusher.await.unwrap();
assert_eq!(q.pop().await, Some(2));
assert_eq!(q.accepted_count(), 2);
}
#[tokio::test]
async fn close_drains_then_returns_none() {
let q: BackpressureQueue<u32> = BackpressureQueue::new(2, OverflowPolicy::Reject);
q.push(1).await.unwrap();
q.push(2).await.unwrap();
let q2 = q.clone();
let drainer = tokio::spawn(async move {
let mut seen = Vec::new();
while let Some(v) = q2.pop().await {
seen.push(v);
}
seen
});
q.close();
assert_eq!(drainer.await.unwrap(), vec![1, 2]);
assert!(q.push(3).await.is_err());
}
}
+365
View File
@@ -0,0 +1,365 @@
//! Circuit breaker for isolating failures in calls to external services
//! (feature `resiliency`).
//!
//! A [`CircuitBreaker`] tracks consecutive failures on calls it guards. Once
//! failures reach a threshold it *trips open*, refusing further calls for a
//! cooldown window so the target gets time to recover. After the window it
//! transitions to *half-open*, admitting a small number of probe calls; a
//! successful probe closes the circuit, a failed probe re-opens it.
use std::sync::atomic::{AtomicU64, AtomicU8, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
/// The current state of a circuit.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CircuitState {
/// Requests flow normally; failures accumulate toward the threshold.
Closed,
/// Requests are refused while the circuit waits to recover.
Open,
/// A limited number of probe requests are admitted to test recovery.
HalfOpen,
}
const CLOSED: u8 = 0;
const OPEN: u8 = 1;
const HALF_OPEN: u8 = 2;
/// Configuration for a [`CircuitBreaker`].
#[derive(Debug, Clone)]
pub struct CircuitBreakerConfig {
/// Consecutive failures that trip `Closed` -> `Open`.
pub failure_threshold: u64,
/// How long to remain `Open` before moving to `HalfOpen`.
pub open_timeout: Duration,
/// Maximum concurrent probe calls admitted while `HalfOpen`.
pub half_open_max_calls: usize,
/// Consecutive successes that close `HalfOpen` -> `Closed`.
pub close_successes: u64,
}
impl Default for CircuitBreakerConfig {
fn default() -> Self {
Self {
failure_threshold: 5,
open_timeout: Duration::from_secs(30),
half_open_max_calls: 1,
close_successes: 1,
}
}
}
/// The result of a guarded call.
#[derive(Debug)]
pub enum CircuitError<E> {
/// The circuit is open and the call was refused without executing.
Open,
/// The guarded operation failed (and the failure was recorded).
Inner(E),
}
impl<E: std::fmt::Display> std::fmt::Display for CircuitError<E> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Open => write!(f, "circuit breaker is open"),
Self::Inner(err) => write!(f, "guarded call failed: {err}"),
}
}
}
impl<E: std::fmt::Debug + std::fmt::Display> std::error::Error for CircuitError<E> {}
struct Inner {
state: AtomicU8,
failures: AtomicU64,
successes: AtomicU64,
half_open_in_flight: AtomicUsize,
opened_at: Mutex<Option<Instant>>,
config: CircuitBreakerConfig,
}
/// A thread-safe circuit breaker.
///
/// Construct with [`CircuitBreaker::new`]; use [`CircuitBreaker::call`] to
/// guard a synchronous operation, or [`CircuitBreaker::allow_request`] +
/// [`CircuitBreaker::record_success`] / [`CircuitBreaker::record_failure`] to
/// guard an async call that cannot hold a borrow across an `.await`.
#[derive(Clone)]
pub struct CircuitBreaker {
inner: Arc<Inner>,
}
impl CircuitBreaker {
/// Builds a new breaker with the given configuration.
pub fn new(config: CircuitBreakerConfig) -> Self {
Self {
inner: Arc::new(Inner {
state: AtomicU8::new(CLOSED),
failures: AtomicU64::new(0),
successes: AtomicU64::new(0),
half_open_in_flight: AtomicUsize::new(0),
opened_at: Mutex::new(None),
config,
}),
}
}
/// Returns the current circuit state, applying the open->half-open
/// transition if the cooldown has elapsed.
pub fn state(&self) -> CircuitState {
match self.inner.state.load(Ordering::Acquire) {
OPEN => {
let opened = self.inner.opened_at.lock().unwrap();
if opened
.map(|t| t.elapsed() >= self.inner.config.open_timeout)
.unwrap_or(false)
{
self.transition(HALF_OPEN);
CircuitState::HalfOpen
} else {
CircuitState::Open
}
}
HALF_OPEN => CircuitState::HalfOpen,
_ => CircuitState::Closed,
}
}
/// Whether requests should currently be refused.
pub fn is_open(&self) -> bool {
self.state() == CircuitState::Open
}
/// Whether a request may proceed right now without tripping the breaker.
///
/// Useful as a gate for an async call: call this before `.await`ing, then
/// report the outcome via [`CircuitBreaker::record_success`] /
/// [`CircuitBreaker::record_failure`].
pub fn allow_request(&self) -> bool {
match self.state() {
CircuitState::Open => false,
CircuitState::HalfOpen => {
let in_flight = self.inner.half_open_in_flight.load(Ordering::Acquire);
if in_flight < self.inner.config.half_open_max_calls {
self.inner
.half_open_in_flight
.fetch_add(1, Ordering::AcqRel);
true
} else {
false
}
}
CircuitState::Closed => true,
}
}
/// Runs `f`, recording its outcome and refusing the call if the circuit is
/// open.
pub fn call<T, E, F>(&self, f: F) -> Result<T, CircuitError<E>>
where
F: FnOnce() -> Result<T, E>,
{
if !self.allow_request() {
return Err(CircuitError::Open);
}
let result = f();
match result {
Ok(value) => {
self.record_result(true);
Ok(value)
}
Err(err) => {
self.record_result(false);
Err(CircuitError::Inner(err))
}
}
}
/// Records that a (previously-admitted) call succeeded.
///
/// Use with [`CircuitBreaker::allow_request`] when guarding an async call.
pub fn record_success(&self) {
self.record_result(true)
}
/// Records that a (previously-admitted) call failed.
///
/// Use with [`CircuitBreaker::allow_request`] when guarding an async call.
pub fn record_failure(&self) {
self.record_result(false)
}
/// Resets the breaker to its initial `Closed` state.
pub fn reset(&self) {
self.inner.state.store(CLOSED, Ordering::Release);
self.inner.failures.store(0, Ordering::Release);
self.inner.successes.store(0, Ordering::Release);
self.inner.half_open_in_flight.store(0, Ordering::Release);
*self.inner.opened_at.lock().unwrap() = None;
}
fn record_result(&self, success: bool) {
match self.inner.state.load(Ordering::Acquire) {
HALF_OPEN => {
// A probe finished; release its slot regardless of outcome.
self.inner
.half_open_in_flight
.fetch_sub(1, Ordering::AcqRel);
if success {
let successes = self.inner.successes.fetch_add(1, Ordering::AcqRel) + 1;
if successes >= self.inner.config.close_successes {
self.close();
}
} else {
self.inner.failures.fetch_add(1, Ordering::AcqRel);
self.open();
}
}
OPEN => {
if success {
self.close();
}
}
_ => {
if success {
self.inner.failures.store(0, Ordering::Release);
} else {
let failures = self.inner.failures.fetch_add(1, Ordering::AcqRel) + 1;
if failures >= self.inner.config.failure_threshold {
self.open();
}
}
}
}
}
fn transition(&self, next: u8) {
self.inner.state.store(next, Ordering::Release);
}
fn close(&self) {
self.inner.state.store(CLOSED, Ordering::Release);
self.inner.failures.store(0, Ordering::Release);
self.inner.successes.store(0, Ordering::Release);
self.inner.half_open_in_flight.store(0, Ordering::Release);
self.inner.opened_at.lock().unwrap().take();
}
fn open(&self) {
self.inner.state.store(OPEN, Ordering::Release);
self.inner.failures.store(0, Ordering::Release);
self.inner.successes.store(0, Ordering::Release);
self.inner.half_open_in_flight.store(0, Ordering::Release);
*self.inner.opened_at.lock().unwrap() = Some(Instant::now());
}
}
#[cfg(test)]
mod tests {
use super::*;
fn breaker() -> CircuitBreaker {
CircuitBreaker::new(CircuitBreakerConfig {
failure_threshold: 3,
open_timeout: Duration::from_millis(50),
half_open_max_calls: 1,
close_successes: 1,
})
}
#[test]
fn closed_start_state() {
let b = breaker();
assert_eq!(b.state(), CircuitState::Closed);
assert!(!b.is_open());
}
#[test]
fn trips_open_after_threshold() {
let b = breaker();
for _ in 0..3 {
let r: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
assert!(r.is_err());
}
assert_eq!(b.state(), CircuitState::Open);
assert!(b.is_open());
}
#[test]
fn open_refuses_calls() {
let b = breaker();
for _ in 0..3 {
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
}
let refused: Result<(), CircuitError<u8>> = b.call(|| Ok(()));
assert!(matches!(refused, Err(CircuitError::Open)));
}
#[test]
fn success_resets_failure_count_in_closed() {
let b = breaker();
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
assert_eq!(b.state(), CircuitState::Closed);
let r: Result<(), CircuitError<u8>> = b.call(|| Ok(()));
assert!(r.is_ok());
// Failure count reset — two more failures should not trip (needs 3).
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
assert_eq!(b.state(), CircuitState::Closed);
}
#[test]
fn half_open_closes_after_success() {
let b = breaker();
for _ in 0..3 {
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
}
assert_eq!(b.state(), CircuitState::Open);
std::thread::sleep(Duration::from_millis(60));
// Once half-open, a success closes the circuit.
let admitted = {
// allow_request true implies half-open probe admitted
b.allow_request()
};
if admitted {
b.record_success();
}
assert_eq!(b.state(), CircuitState::Closed);
}
#[test]
fn half_open_failure_reopens() {
let b = breaker();
for _ in 0..3 {
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
}
std::thread::sleep(Duration::from_millis(60));
if b.allow_request() {
b.record_failure();
}
assert_eq!(b.state(), CircuitState::Open);
}
#[test]
fn reset_returns_to_closed() {
let b = breaker();
for _ in 0..3 {
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
}
assert_eq!(b.state(), CircuitState::Open);
b.reset();
assert_eq!(b.state(), CircuitState::Closed);
}
#[test]
fn tracks_inner_error_and_success() {
let b = breaker();
let ok: Result<u32, CircuitError<u8>> = b.call(|| Ok(5u32));
assert_eq!(ok.unwrap(), 5);
let err: Result<u32, CircuitError<u8>> = b.call(|| Err(9u8));
assert!(matches!(err, Err(CircuitError::Inner(9))));
}
}
+6 -6
View File
@@ -3,12 +3,12 @@
use std::panic::{catch_unwind, AssertUnwindSafe};
use crate::context::context;
use crate::error::CorexError;
use crate::error::MytheclipseError;
/// Runs `f` on the global compute thread pool and returns its result.
///
/// If `f` panics, the panic is caught and converted into
/// [`CorexError::ComputePanic`] instead of unwinding across the pool
/// [`MytheclipseError::ComputePanic`] instead of unwinding across the pool
/// boundary or poisoning the pool; subsequent calls to [`compute`] continue
/// to work normally.
///
@@ -23,7 +23,7 @@ use crate::error::CorexError;
/// captured by mutable reference: if `f` panics partway through mutating a
/// captured `&mut T`, that `T` may be left in an inconsistent state from
/// the caller's perspective.
pub fn compute<F, R>(f: F) -> Result<R, CorexError>
pub fn compute<F, R>(f: F) -> Result<R, MytheclipseError>
where
F: FnOnce() -> R + Send,
R: Send,
@@ -32,7 +32,7 @@ where
context()
.compute_pool
.install(move || catch_unwind(wrapped))
.map_err(|payload| CorexError::ComputePanic(panic_payload_to_string(payload)))
.map_err(|payload| MytheclipseError::ComputePanic(panic_payload_to_string(payload)))
}
fn panic_payload_to_string(payload: Box<dyn std::any::Any + Send>) -> String {
@@ -51,8 +51,8 @@ mod tests {
#[test]
fn compute_panic_is_isolated_and_pool_survives() {
let panicked: Result<u32, CorexError> = compute(|| panic!("boom"));
assert!(matches!(panicked, Err(CorexError::ComputePanic(_))));
let panicked: Result<u32, MytheclipseError> = compute(|| panic!("boom"));
assert!(matches!(panicked, Err(MytheclipseError::ComputePanic(_))));
let recovered = compute(|| 1 + 1);
assert_eq!(recovered.unwrap(), 2);
+180
View File
@@ -0,0 +1,180 @@
//! A synchronous concurrency limiter (feature `traffic`).
//!
//! [`ConcurrencyLimiter`] bounds how many threads may hold a permit
//! simultaneously, providing a way to cap concurrent access to an expensive
//! resource (e.g. at most 10 tasks may hold a database connection or compress
//! an image at once) without requiring a runtime. It is built on std's
//! `Mutex` + `Condvar` and therefore works in plain, non-async threads.
use std::sync::Arc;
use std::sync::{Condvar, Mutex};
struct Inner {
state: Mutex<usize>,
available: Condvar,
max: usize,
}
/// A thread-safe cap on concurrent in-flight sections.
#[derive(Clone)]
pub struct ConcurrencyLimiter {
inner: Arc<Inner>,
}
/// An RAII guard holding a concurrency permit.
///
/// Released (the permit returning to the limiter) when this guard is dropped.
#[must_use = "dropping the permit releases the slot; if the caller wants to hold it, keep it alive"]
pub struct ConcurrencyPermit {
inner: Arc<Inner>,
}
impl ConcurrencyLimiter {
/// Builds a limiter allowing at most `max_concurrent` held permits.
///
/// # Panics
///
/// Panics if `max_concurrent` is zero.
pub fn new(max_concurrent: usize) -> Self {
assert!(max_concurrent > 0, "concurrency limit must be > 0");
Self {
inner: Arc::new(Inner {
state: Mutex::new(max_concurrent),
available: Condvar::new(),
max: max_concurrent,
}),
}
}
/// Blocks the calling thread until a permit is available, then returns it.
pub fn acquire(&self) -> ConcurrencyPermit {
let mut available = self.inner.state.lock().unwrap();
while *available == 0 {
available = self
.inner
.available
.wait(available)
.expect("concurrency limiter condvar poisoned");
}
*available -= 1;
ConcurrencyPermit {
inner: Arc::clone(&self.inner),
}
}
/// Attempts to acquire a permit without blocking.
///
/// Returns `None` if the limiter is currently at its maximum.
pub fn try_acquire(&self) -> Option<ConcurrencyPermit> {
let mut available = self.inner.state.lock().unwrap();
if *available == 0 {
return None;
}
*available -= 1;
Some(ConcurrencyPermit {
inner: Arc::clone(&self.inner),
})
}
/// How many permits are currently held.
pub fn in_use(&self) -> usize {
let available = *self.inner.state.lock().unwrap();
self.inner.max - available
}
/// The maximum number of concurrently held permits.
pub fn max(&self) -> usize {
self.inner.max
}
}
impl Drop for ConcurrencyPermit {
fn drop(&mut self) {
let mut available = self.inner.state.lock().unwrap();
*available += 1;
self.inner.available.notify_one();
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
#[test]
fn acquires_and_releases() {
let limiter = ConcurrencyLimiter::new(1);
{
let _permit = limiter.try_acquire().expect("should acquire");
assert_eq!(limiter.in_use(), 1);
}
assert_eq!(limiter.in_use(), 0);
}
#[test]
fn try_acquire_fails_at_cap() {
let limiter = ConcurrencyLimiter::new(2);
let _a = limiter.try_acquire().expect("first");
let _b = limiter.try_acquire().expect("second");
assert!(limiter.try_acquire().is_none());
drop(_a);
assert!(limiter.try_acquire().is_some());
}
#[test]
fn acquire_blocks_until_slot_frees() {
let limiter = ConcurrencyLimiter::new(1);
let permit = limiter.try_acquire().unwrap();
let in_flight = Arc::new(AtomicUsize::new(0));
let max_observed = Arc::new(AtomicUsize::new(0));
let limiter2 = limiter.clone();
let in_flight2 = Arc::clone(&in_flight);
let max2 = Arc::clone(&max_observed);
let thread = std::thread::spawn(move || {
let _permit = limiter2.acquire();
in_flight2.fetch_add(1, Ordering::SeqCst);
max2.fetch_max(in_flight2.load(Ordering::SeqCst), Ordering::SeqCst);
});
std::thread::sleep(Duration::from_millis(40));
assert_eq!(in_flight.load(Ordering::SeqCst), 0);
// Explicitly drop permit so waiting thread unblocks
drop(permit);
thread.join().unwrap();
assert_eq!(max_observed.load(Ordering::SeqCst), 1);
assert_eq!(in_flight.load(Ordering::SeqCst), 1);
}
#[test]
fn limits_concurrent_sections() {
let limiter = ConcurrencyLimiter::new(2);
let in_flight = Arc::new(AtomicUsize::new(0));
let max_observed = Arc::new(AtomicUsize::new(0));
let mut threads = Vec::new();
for _ in 0..8 {
let limiter = limiter.clone();
let in_flight = Arc::clone(&in_flight);
let max_observed = Arc::clone(&max_observed);
threads.push(std::thread::spawn(move || {
let _permit = limiter.acquire();
in_flight.fetch_add(1, Ordering::SeqCst);
max_observed.fetch_max(in_flight.load(Ordering::SeqCst), Ordering::SeqCst);
std::thread::sleep(Duration::from_millis(10));
in_flight.fetch_sub(1, Ordering::SeqCst);
}));
}
for t in threads {
t.join().unwrap();
}
assert!(max_observed.load(Ordering::SeqCst) <= 2);
}
#[test]
#[should_panic]
fn zero_limit_panics() {
let _ = ConcurrencyLimiter::new(0);
}
}
+475
View File
@@ -0,0 +1,475 @@
//! Minimal cron scheduling (feature `lifecycle`).
//!
//! [`CronSchedule`] parses the standard five-field cron expression
//! (`minute hour day-of-month month day-of-week`) and computes next fire
//! times. [`schedule`] drives a [`CronSchedule`] on a Tokio runtime, running a
//! job each time it fires. The implementation is self-contained — no external
//! cron crate — and deliberately ignores time zones (dates are interpreted in
//! UTC).
use std::time::Duration;
use tracing::Instrument;
/// Error parsing a cron expression.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CronParseError(pub String);
impl std::fmt::Display for CronParseError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "invalid cron expression: {}", self.0)
}
}
impl std::error::Error for CronParseError {}
/// Error scheduling a cron job.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CronError {
/// The expression did not parse.
Parse(CronParseError),
}
impl std::fmt::Display for CronError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Parse(err) => write!(f, "cron error: {err}"),
}
}
}
impl std::error::Error for CronError {}
/// A simple wall-clock instant used by cron computations.
///
/// Fields follow the chronological order used in cron expressions, resolved to
/// the local (UTC) civil calendar.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct CronTime {
/// Year, e.g. `2026`.
pub year: i32,
/// Month, 1-12.
pub month: u32,
/// Day of month, 1-31.
pub day: u32,
/// Hour, 0-23.
pub hour: u32,
/// Minute, 0-59.
pub minute: u32,
/// Second, 0-59 (always 0 for scheduled fires).
pub second: u32,
}
/// A parsed cron schedule.
#[derive(Debug, Clone)]
pub struct CronSchedule {
minutes: Vec<u8>,
hours: Vec<u8>,
days_of_month: Vec<u8>,
months: Vec<u8>,
days_of_week: Vec<u8>,
}
/// A handle to a running cron job; dropping it aborts the task.
#[derive(Debug)]
pub struct CronJob {
handle: tokio::task::JoinHandle<()>,
}
impl CronJob {
/// Cancels the scheduled job.
pub fn cancel(self) {
self.handle.abort();
}
}
impl CronSchedule {
/// Parses the standard five-field cron expression
/// `minute hour day-of-month month day-of-week`.
///
/// Supported tokens per field: `*`, `*/step`, a literal, a list
/// (`a,b,c`), an inclusive range (`a-b`), and a ranged step (`a-b/step`).
/// Day-of-week also accepts `SUN`..`SAT` (Sun=0) names. Day-of-month and
/// day-of-week are combined with an OR when both are restricted, matching
/// standard cron semantics.
pub fn parse(expr: &str) -> Result<Self, CronParseError> {
let fields: Vec<&str> = expr.split_whitespace().collect();
if fields.len() != 5 {
return Err(CronParseError(format!(
"expected exactly 5 fields, got {}",
fields.len()
)));
}
let minutes = parse_field(fields[0], 0, 59, false)?;
let hours = parse_field(fields[1], 0, 23, false)?;
let days_of_month = parse_field(fields[2], 1, 31, false)?;
let months = parse_field(fields[3], 1, 12, false)?;
let days_of_week = parse_field(fields[4], 0, 6, true)?;
Ok(Self {
minutes,
hours,
days_of_month,
months,
days_of_week,
})
}
/// Returns the next fire time strictly after `from`, or `None` if the
/// expression cannot fire within the scan horizon (e.g. Feb 30).
///
/// Scans forward minute-by-minute up to roughly five years.
pub fn next_after(&self, from: CronTime) -> Option<CronTime> {
// Scan whole minutes (second = 0), starting strictly after `from`.
let mut act = from.advance_minute();
let horizon = 5 * 366 * 24 * 60; // minutes in ~5 years
for _ in 0..horizon {
if self.matches(act) {
return Some(act);
}
act = act.advance_minute();
}
None
}
fn matches(&self, t: CronTime) -> bool {
let month_ok = self.months.contains(&(t.month as u8));
let dom_ok = self.days_of_month.contains(&(t.day as u8));
let dow_ok = self.days_of_week.contains(&(weekday(t) as u8));
// Standard cron: if both dom and dow are restricted, a match on EITHER
// is sufficient; if only one is restricted, it must match; if neither,
// always true.
let dom_restricted = self.days_of_month.len() < 31;
let dow_restricted = self.days_of_week.len() < 7;
let day_ok = match (dom_restricted, dow_restricted) {
(true, true) => dom_ok || dow_ok,
(true, false) => dom_ok,
(false, true) => dow_ok,
(false, false) => true,
};
month_ok
&& day_ok
&& self.hours.contains(&(t.hour as u8))
&& self.minutes.contains(&(t.minute as u8))
}
/// The next several fire times, for display/tests.
pub fn next_five(&self, from: CronTime) -> Vec<CronTime> {
let mut out = Vec::new();
let mut cur = from;
for _ in 0..5 {
match self.next_after(cur) {
Some(next) => {
out.push(next);
cur = next;
}
None => break,
}
}
out
}
}
/// Spawns a task that runs `job` on `expr`'s schedule until cancelled.
///
/// The job is a closure returning a future; each invocation runs in a
/// `mytheclipse_cron_task` tracing span. A fire time that was missed while the
/// job was running is not backlogged — the next scheduled fire is used.
pub fn schedule<F, Fut>(expr: &str, mut job: F) -> Result<CronJob, CronError>
where
F: FnMut() -> Fut + Send + 'static,
Fut: std::future::Future<Output = ()> + Send + 'static,
{
let schedule = CronSchedule::parse(expr).map_err(CronError::Parse)?;
let expr_owned = expr.to_string();
let handle = tokio::spawn(async move {
loop {
let now = CronTime::now();
let next = schedule.next_after(now);
let next = match next {
Some(next) => next,
None => {
// Unreachable schedule; wait and re-check periodically.
tokio::time::sleep(Duration::from_secs(60)).await;
continue;
}
};
let now_ts = CronTime::now().to_timestamp();
let until = tokio::time::Instant::now()
+ Duration::from_secs(next.to_timestamp().saturating_sub(now_ts).max(1) as u64);
tokio::time::sleep_until(until).await;
let span = tracing::info_span!(
"mytheclipse_cron_task",
expr = %expr_owned
);
job().instrument(span).await;
}
});
Ok(CronJob { handle })
}
impl CronTime {
/// The current wall-clock time (UTC), truncated to minutes for firing.
pub fn now() -> Self {
// std has no direct civil-date conversion; derive from SystemTime.
now_utc()
}
fn to_timestamp(self) -> i64 {
days_from_civil(self.year, self.month, self.day) * 86_400
+ (self.hour as i64) * 3_600
+ (self.minute as i64) * 60
+ self.second as i64
}
fn advance_minute(self) -> Self {
let mut t = self.add_seconds(60);
t.second = 0;
t
}
fn add_seconds(self, secs: i64) -> Self {
let ts = self.to_timestamp() + secs;
from_timestamp(ts)
}
}
/// Weekday: 0 = Sunday .. 6 = Saturday (matches cron).
fn weekday(t: CronTime) -> u32 {
// 1970-01-01 was a Thursday (weekday 4).
let days = days_from_civil(t.year, t.month, t.day);
// ((days + 4) mod 7) => 0=Sunday
((days % 7 + 11) % 7) as u32
}
/// Days since 1970-01-01 for a civil date (Howard Hinnant's algorithm).
fn days_from_civil(y: i32, m: u32, d: u32) -> i64 {
let y = y as i64 - (m as i64 <= 2) as i64;
let era = (if y >= 0 { y } else { y - 399 }) / 400;
let yoe = y - era * 400; // [0, 399]
let mp = (m as i64 + 9) % 12; // March=0
let doy = (153 * mp + 2) / 5 + d as i64 - 1;
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
era * 146_097 + doe - 719_468
}
/// Converts a Unix timestamp (seconds) to a civil [`CronTime`].
fn from_timestamp(ts: i64) -> CronTime {
let days = ts.div_euclid(86_400);
let z = days + 719_468; // days since civil epoch
let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
let doe = z - era * 146_097; // [0, 146096]
let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
let y = yoe + era * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 } as u32;
let y = if m <= 2 { y + 1 } else { y } as i32;
let (h, mi, s) = split_seconds(ts);
CronTime {
year: y,
month: m,
day: d as u32,
hour: h,
minute: mi,
second: s,
}
}
fn split_seconds(ts: i64) -> (u32, u32, u32) {
let rem = ts.rem_euclid(86_400);
(
(rem / 3_600) as u32,
((rem % 3_600) / 60) as u32,
(rem % 60) as u32,
)
}
/// Returns the current UTC civil time plus second accuracy from SystemTime.
fn now_utc() -> CronTime {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default();
from_timestamp(now.as_secs() as i64)
}
/// Parses a single cron field into a sorted list of allowed values.
fn parse_field(
field: &str,
min: u8,
max: u8,
allow_names: bool,
) -> Result<Vec<u8>, CronParseError> {
let mut out = Vec::new();
for part in field.split(',') {
let part = part.trim();
if part.is_empty() {
return Err(CronParseError(format!("empty field element in `{field}`")));
}
// Parse `base` optionally followed by `/step`.
let (range, step) = match part.split_once('/') {
Some((r, s)) => (r, parse_num(s, "step")?),
None => (part, 1),
};
// Parse `*`, `a-b`, or literal.
match range {
"*" => {
let mut v = min;
while v <= max {
out.push(v);
v += step;
}
}
_ if range.contains('-') => {
let (a, b) = range
.split_once('-')
.ok_or_else(|| CronParseError(format!("bad range `{range}`")))?;
let a = parse_value(a, min, max, allow_names)?;
let b = parse_value(b, min, max, allow_names)?;
if a > b {
return Err(CronParseError(format!("range start {a} > end {b}")));
}
let mut v = a;
while v <= b {
out.push(v);
v += step;
}
}
_ => {
let v = parse_value(range, min, max, allow_names)?;
out.push(v);
}
}
}
out.sort_unstable();
out.dedup();
Ok(out)
}
fn parse_num(s: &str, what: &str) -> Result<u8, CronParseError> {
s.parse::<u8>()
.map_err(|_| CronParseError(format!("invalid {what} `{s}`")))
}
fn parse_value(s: &str, min: u8, max: u8, allow_names: bool) -> Result<u8, CronParseError> {
let upper = s.to_uppercase();
// Day-of-week names.
if allow_names {
let name = match upper.as_str() {
"SUN" => Some(0),
"MON" => Some(1),
"TUE" => Some(2),
"WED" => Some(3),
"THU" => Some(4),
"FRI" => Some(5),
"SAT" => Some(6),
_ => None,
};
if let Some(v) = name {
return Ok(v);
}
}
let v = parse_num(s, "value")?;
if v < min || v > max {
return Err(CronParseError(format!(
"value {v} out of range {min}..{max}"
)));
}
Ok(v)
}
#[cfg(test)]
mod tests {
use super::*;
fn t(y: i32, m: u32, d: u32, h: u32, mi: u32) -> CronTime {
CronTime {
year: y,
month: m,
day: d,
hour: h,
minute: mi,
second: 0,
}
}
#[test]
fn parses_star() {
let s = CronSchedule::parse("* * * * *").unwrap();
assert_eq!(s.minutes.len(), 60);
}
#[test]
fn parses_list_and_range() {
let s = CronSchedule::parse("1,15 9-17 * * 1-5").unwrap();
assert_eq!(s.minutes, vec![1, 15]);
assert_eq!(s.hours, (9..=17).collect::<Vec<_>>());
assert_eq!(s.days_of_week, vec![1, 2, 3, 4, 5]);
}
#[test]
fn parses_step_and_names() {
let s = CronSchedule::parse("*/15 * * * MON,WED").unwrap();
assert_eq!(s.minutes, vec![0, 15, 30, 45]);
assert_eq!(s.days_of_week, vec![1, 3]);
}
#[test]
fn rejects_too_few_fields() {
assert!(CronSchedule::parse("* * * *").is_err());
}
#[test]
fn rejects_out_of_range() {
assert!(CronSchedule::parse("60 * * * *").is_err());
}
#[test]
fn next_after_every_minute() {
let s = CronSchedule::parse("* * * * *").unwrap();
let next = s.next_after(t(2026, 8, 28, 10, 30)).unwrap();
assert_eq!(next, t(2026, 8, 28, 10, 31));
}
#[test]
fn next_after_midnight() {
let s = CronSchedule::parse("0 0 * * *").unwrap();
let next = s.next_after(t(2026, 8, 28, 23, 59)).unwrap();
assert_eq!(next, t(2026, 8, 29, 0, 0));
}
#[test]
fn next_after_weekday_only() {
// 2026-08-28 is a Friday (5). "0 9 * * MON-FRI" fires next Monday.
let s = CronSchedule::parse("0 9 * * MON-FRI").unwrap();
let next = s.next_after(t(2026, 8, 28, 23, 0)).unwrap();
// Friday 23:00 -> next is Monday 09:00 (Aug 31).
assert_eq!(next, t(2026, 8, 31, 9, 0));
}
#[test]
fn next_after_leap_year_feb() {
let s = CronSchedule::parse("0 12 29 2 *").unwrap();
// Next Feb 29 after 2026 is 2028.
let next = s.next_after(t(2026, 1, 1, 0, 0)).unwrap();
assert_eq!(next, t(2028, 2, 29, 12, 0));
}
#[test]
fn impossible_expression_returns_none() {
let s = CronSchedule::parse("0 0 30 2 *").unwrap(); // Feb 30 never exists
assert!(s.next_after(t(2026, 1, 1, 0, 0)).is_none());
}
#[test]
fn next_five_every_hour() {
let s = CronSchedule::parse("0 * * * *").unwrap();
let five = s.next_five(t(2026, 8, 28, 9, 45));
assert_eq!(five.len(), 5);
assert_eq!(five[0], t(2026, 8, 28, 10, 0));
assert_eq!(five[1], t(2026, 8, 28, 11, 0));
assert_eq!(five[4], t(2026, 8, 28, 14, 0));
}
}
+45 -5
View File
@@ -1,28 +1,68 @@
//! Shared error type for mytheclipse execution primitives.
/// Errors surfaced by mytheclipse's panic-isolated execution primitives.
/// Errors surfaced by mytheclipse's execution primitives.
///
/// Marked `#[non_exhaustive]` so new variants can be added without a
/// breaking change; downstream `match` expressions should include a
/// wildcard arm.
#[non_exhaustive]
#[derive(Debug)]
pub enum CorexError {
#[derive(Debug, PartialEq, Eq)]
pub enum MytheclipseError {
/// A closure submitted to [`crate::compute::compute`] panicked.
///
/// The contained string is a best-effort rendering of the panic
/// payload; the compute thread pool itself remains usable afterward.
ComputePanic(String),
/// A deadline elapsed before the wrapped future completed
/// ([`crate::timeout::with_timeout`]).
Timeout,
/// A request was refused because the circuit breaker is open
/// ([`crate::circuit_breaker`]).
CircuitOpen,
/// A retry loop gave up after exhausting its maximum attempts
/// ([`crate::retry()`]). The contained count is the number of attempts made.
RetryExhausted { attempts: u32 },
/// A request was refused because the rate limiter had no tokens left
/// ([`crate::ratelimit`]).
RateLimited,
/// A queue was full and the configured overflow policy rejected the item
/// ([`crate::backpressure`]).
QueueFull(String),
/// A concurrency limiter was already at its maximum and refused a request
/// ([`crate::concurrency`]).
ConcurrencyLimitExceeded,
/// Shutdown has been requested for the process or manager
/// ([`crate::shutdown`]).
Shutdown,
/// An observability operation failed, e.g. metric export
/// ([`crate::metrics`]).
Metrics(String),
/// Invalid configuration or input, e.g. a malformed cron expression
/// ([`crate::cron`]).
Config(String),
}
impl std::fmt::Display for CorexError {
impl std::fmt::Display for MytheclipseError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::ComputePanic(message) => {
write!(f, "compute closure panicked: {message}")
}
Self::Timeout => write!(f, "operation timed out"),
Self::CircuitOpen => write!(f, "circuit breaker is open"),
Self::RetryExhausted { attempts } => {
write!(f, "retry exhausted after {attempts} attempts")
}
Self::RateLimited => write!(f, "rate limit exceeded"),
Self::QueueFull(detail) => write!(f, "queue is full: {detail}"),
Self::ConcurrencyLimitExceeded => {
write!(f, "concurrency limit exceeded")
}
Self::Shutdown => write!(f, "shutdown requested"),
Self::Metrics(detail) => write!(f, "metrics error: {detail}"),
Self::Config(detail) => write!(f, "configuration error: {detail}"),
}
}
}
impl std::error::Error for CorexError {}
impl std::error::Error for MytheclipseError {}
+74 -8
View File
@@ -1,8 +1,9 @@
//! # mytheclipse
//!
//! Resource-aware abstractions for async I/O, heavy compute, and
//! background queue management, built on a single lazily-initialized
//! global engine context.
//! Resource-aware abstractions for async I/O, heavy compute, background queue
//! management, resiliency, traffic control, lifecycle, and observability —
//! built on a single lazily-initialized global engine context plus a set of
//! self-contained, constructible utilities.
//!
//! Call [`init`] once at startup (or simply let the first call to any
//! entry point below trigger it lazily) and then use whichever of the
@@ -11,11 +12,18 @@
//! - [`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()`] / [`circuit_breaker`] / [`timeout()`] (feature `resiliency`) — fault tolerance.
//! - [`ratelimit`] / [`backpressure`] / [`concurrency`] (feature `traffic`) — load control.
//! - [`shutdown`] / [`cron`] (feature `lifecycle`) — lifecycle management.
//! - [`metrics`] / [`panic_tracker`] (feature `observability`) — runtime visibility.
//!
//! Enable the `full` feature to pull in all three at once.
//! Enable the `full` feature to pull in all of the above at once. The three
//! execution primitives are sized from the host's logical core count via the
//! engine context; the resiliency/traffic/lifecycle/observability utilities
//! are self-contained and constructed explicitly (e.g.
//! `RateLimiter::new(...)`, `CircuitBreaker::new(...)`).
pub mod context;
#[cfg(feature = "compute")]
pub mod error;
#[cfg(feature = "io")]
@@ -27,10 +35,38 @@ pub mod compute;
#[cfg(feature = "bg")]
pub mod bg;
pub use context::{context, EngineContext};
#[cfg(feature = "resiliency")]
pub mod retry;
#[cfg(feature = "compute")]
pub use error::CorexError;
#[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 = "lifecycle")]
pub mod shutdown;
#[cfg(feature = "lifecycle")]
pub mod cron;
#[cfg(feature = "observability")]
pub mod metrics;
#[cfg(feature = "observability")]
pub mod panic_tracker;
pub use context::{context, EngineContext};
pub use error::MytheclipseError;
#[cfg(feature = "io")]
pub use io::spawn_io;
@@ -41,6 +77,36 @@ pub use compute::compute;
#[cfg(feature = "bg")]
pub use bg::spawn_bg;
#[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 timeout::{timeout, with_timeout, Timeout, TimeoutError};
#[cfg(feature = "traffic")]
pub use ratelimit::{RateLimitError, RateLimiter};
#[cfg(feature = "traffic")]
pub use backpressure::{BackpressureError, BackpressureQueue, OverflowPolicy};
#[cfg(feature = "traffic")]
pub use concurrency::{ConcurrencyLimiter, ConcurrencyPermit};
#[cfg(feature = "lifecycle")]
pub use shutdown::{ShutdownManager, ShutdownSignal};
#[cfg(feature = "lifecycle")]
pub use cron::{schedule, CronError, CronJob, CronParseError, CronSchedule};
#[cfg(feature = "observability")]
pub use metrics::{MetricsCollector, MetricsSnapshot};
#[cfg(feature = "observability")]
pub use panic_tracker::{PanicGuard, PanicInfo, PanicTracker};
/// Bootstraps the global [`EngineContext`].
///
/// See [`context::init`] for full semantics: this is safe to call any
+337
View File
@@ -0,0 +1,337 @@
//! A centralized, thread-safe metrics collector (feature `observability`).
//!
//! [`MetricsCollector`] records runtime health statistics — active threads,
//! remaining queue capacity, average task duration, custom counters and gauges
//! — and can render them in the [Prometheus text exposition format](https://prometheus.io/docs/instrumenting/exposition_formats/)
//! for a scraper, or as a structured [`MetricsSnapshot`] that a downstream adapter
//! can forward (e.g. to OpenTelemetry).
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
/// A single histogram bucket: accumulates `count` observations and their
/// total duration for an average.
#[derive(Debug, Clone, Default)]
struct Histogram {
count: u64,
total_micros: u128,
}
#[derive(Default)]
struct Inner {
active_threads: AtomicUsize,
queue_capacity_total: AtomicUsize,
queue_capacity_remaining: AtomicUsize,
task_count: AtomicU64,
total_task_micros: AtomicU64,
counters: Mutex<HashMap<String, u64>>,
gauges: Mutex<HashMap<String, f64>>,
histograms: Mutex<HashMap<String, Histogram>>,
}
/// A thread-safe handle to a metrics collector.
///
/// Cheap to clone (shared state). Collects data updated from any thread and
/// exports it on demand.
#[derive(Clone, Default)]
pub struct MetricsCollector {
inner: Arc<Inner>,
}
impl MetricsCollector {
/// Builds an empty collector.
pub fn new() -> Self {
Self::default()
}
/// Records that a task ran for `duration`, updating the running task
/// count and average.
pub fn record_task(&self, duration: Duration) {
self.inner.task_count.fetch_add(1, Ordering::AcqRel);
self.inner
.total_task_micros
.fetch_add(duration.as_micros() as u64, Ordering::AcqRel);
}
/// Sets the observed number of active worker threads.
pub fn set_active_threads(&self, n: usize) {
self.inner.active_threads.store(n, Ordering::Release);
}
/// Sets the queue's total capacity.
pub fn set_queue_capacity(&self, n: usize) {
self.inner.queue_capacity_total.store(n, Ordering::Release);
}
/// Sets the queue's currently remaining capacity.
pub fn set_queue_remaining(&self, n: usize) {
self.inner
.queue_capacity_remaining
.store(n, Ordering::Release);
}
/// Increments a counter by `by`.
pub fn inc_counter(&self, name: &str, by: u64) {
*self
.inner
.counters
.lock()
.unwrap()
.entry(name.to_string())
.or_insert(0) += by;
}
/// Sets a gauge to `value`.
pub fn set_gauge(&self, name: &str, value: f64) {
self.inner
.gauges
.lock()
.unwrap()
.insert(name.to_string(), value);
}
/// Records `duration` into the histogram named `name`.
pub fn observe(&self, name: &str, duration: Duration) {
let mut histos = self.inner.histograms.lock().unwrap();
let h = histos.entry(name.to_string()).or_default();
h.count += 1;
h.total_micros += duration.as_micros();
}
/// Total number of recorded tasks.
pub fn task_count(&self) -> u64 {
self.inner.task_count.load(Ordering::Acquire)
}
/// Average task duration, if any tasks have been recorded.
pub fn avg_task_duration(&self) -> Option<Duration> {
let count = self.task_count();
if count == 0 {
return None;
}
let total = self.inner.total_task_micros.load(Ordering::Acquire);
Some(Duration::from_micros(total / count))
}
/// Number of active worker threads last recorded.
pub fn active_threads(&self) -> usize {
self.inner.active_threads.load(Ordering::Acquire)
}
/// Takes a structured snapshot of the collector's state.
pub fn snapshot(&self) -> MetricsSnapshot {
MetricsSnapshot {
active_threads: self.active_threads(),
queue_capacity_total: self.inner.queue_capacity_total.load(Ordering::Acquire),
queue_capacity_remaining: self.inner.queue_capacity_remaining.load(Ordering::Acquire),
task_count: self.task_count(),
avg_task_duration_micros: self.avg_task_duration().map(|d| d.as_micros() as u64),
counters: self
.inner
.counters
.lock()
.unwrap()
.iter()
.map(|(k, v)| (k.clone(), *v))
.collect(),
gauges: self
.inner
.gauges
.lock()
.unwrap()
.iter()
.map(|(k, v)| (k.clone(), *v))
.collect(),
histograms: self
.inner
.histograms
.lock()
.unwrap()
.iter()
.map(|(k, h)| {
(
k.clone(),
HistogramSnapshot {
count: h.count,
avg_micros: if h.count > 0 {
Some((h.total_micros / h.count as u128) as u64)
} else {
None
},
},
)
})
.collect(),
}
}
/// Renders the collector in the Prometheus text exposition format.
///
/// The output is suitable to serve with
/// `Content-Type: text/plain; version=0.0.4`.
pub fn export_prometheus(&self) -> String {
let mut out = String::new();
let snap = self.snapshot();
out.push_str("# HELP mytheclipse_active_threads Number of active worker threads.\n");
out.push_str("# TYPE mytheclipse_active_threads gauge\n");
out.push_str(&format!(
"mytheclipse_active_threads {}\n",
snap.active_threads
));
out.push_str("# HELP mytheclipse_queue_capacity_total Total queue capacity.\n");
out.push_str("# TYPE mytheclipse_queue_capacity_total gauge\n");
out.push_str(&format!(
"mytheclipse_queue_capacity_total {}\n",
snap.queue_capacity_total
));
out.push_str("# HELP mytheclipse_queue_capacity_remaining Remaining queue capacity.\n");
out.push_str("# TYPE mytheclipse_queue_capacity_remaining gauge\n");
out.push_str(&format!(
"mytheclipse_queue_capacity_remaining {}\n",
snap.queue_capacity_remaining
));
out.push_str("# HELP mytheclipse_task_count Total tasks recorded.\n");
out.push_str("# TYPE mytheclipse_task_count counter\n");
out.push_str(&format!("mytheclipse_task_count {}\n", snap.task_count));
if let Some(avg) = snap.avg_task_duration_micros {
out.push_str(
"# HELP mytheclipse_task_duration_avg Average task duration in microseconds.\n",
);
out.push_str("# TYPE mytheclipse_task_duration_avg gauge\n");
out.push_str(&format!("mytheclipse_task_duration_avg {avg}\n"));
}
let mut counters: Vec<_> = snap.counters.into_iter().collect();
counters.sort_by(|a, b| a.0.cmp(&b.0));
for (name, value) in counters {
out.push_str(&format!("# TYPE {name} counter\n"));
out.push_str(&format!("{name} {value}\n"));
}
let mut gauges: Vec<_> = snap.gauges.into_iter().collect();
gauges.sort_by(|a, b| a.0.cmp(&b.0));
for (name, value) in gauges {
out.push_str(&format!("# TYPE {name} gauge\n"));
out.push_str(&format!("{name} {value}\n"));
}
let mut histos: Vec<_> = snap.histograms.into_iter().collect();
histos.sort_by(|a, b| a.0.cmp(&b.0));
for (name, h) in histos {
if let Some(avg) = h.avg_micros {
out.push_str(&format!("# TYPE {name}_count counter\n"));
out.push_str(&format!("{name}_count {}\n", h.count));
out.push_str(&format!("# TYPE {name}_avg gauge\n"));
out.push_str(&format!("{name}_avg {avg}\n"));
}
}
out
}
}
/// A structured, serializable view of a [`MetricsCollector`], suitable for
/// forwarding to OpenTelemetry or another backend.
#[derive(Debug, Clone)]
pub struct MetricsSnapshot {
/// Number of active worker threads last recorded.
pub active_threads: usize,
/// Total queue capacity last recorded.
pub queue_capacity_total: usize,
/// Remaining queue capacity last recorded.
pub queue_capacity_remaining: usize,
/// Total tasks recorded.
pub task_count: u64,
/// Average task duration in microseconds, if any.
pub avg_task_duration_micros: Option<u64>,
/// Named counters.
pub counters: HashMap<String, u64>,
/// Named gauges.
pub gauges: HashMap<String, f64>,
/// Named histogram aggregates.
pub histograms: HashMap<String, HistogramSnapshot>,
}
/// Aggregated view of one histogram.
#[derive(Debug, Clone)]
pub struct HistogramSnapshot {
/// Number of observations.
pub count: u64,
/// Average observation value in microseconds, if any.
pub avg_micros: Option<u64>,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn records_task_and_average() {
let m = MetricsCollector::new();
assert_eq!(m.task_count(), 0);
assert!(m.avg_task_duration().is_none());
m.record_task(Duration::from_millis(10));
m.record_task(Duration::from_millis(20));
assert_eq!(m.task_count(), 2);
assert_eq!(m.avg_task_duration().unwrap(), Duration::from_millis(15));
}
#[test]
fn tracks_threads_and_queue() {
let m = MetricsCollector::new();
m.set_active_threads(8);
m.set_queue_capacity(100);
m.set_queue_remaining(42);
let snap = m.snapshot();
assert_eq!(snap.active_threads, 8);
assert_eq!(snap.queue_capacity_total, 100);
assert_eq!(snap.queue_capacity_remaining, 42);
}
#[test]
fn custom_counters_and_gauges() {
let m = MetricsCollector::new();
m.inc_counter("reqs", 3);
m.inc_counter("reqs", 2);
m.set_gauge("temp", 21.5);
let snap = m.snapshot();
assert_eq!(snap.counters["reqs"], 5);
assert_eq!(snap.gauges["temp"], 21.5);
}
#[test]
fn observe_accumulates_histogram() {
let m = MetricsCollector::new();
m.observe("latency", Duration::from_millis(100));
m.observe("latency", Duration::from_millis(300));
let snap = m.snapshot();
let h = &snap.histograms["latency"];
assert_eq!(h.count, 2);
assert_eq!(h.avg_micros.unwrap(), 200_000);
}
#[test]
fn prometheus_export_contains_lines() {
let m = MetricsCollector::new();
m.record_task(Duration::from_millis(5));
m.set_active_threads(4);
m.inc_counter("my_reqs", 7);
m.set_gauge("my_temp", 1.5);
m.observe("my_lat_ms", Duration::from_millis(12));
let out = m.export_prometheus();
assert!(out.contains("# TYPE mytheclipse_task_count counter"));
assert!(out.contains("mytheclipse_task_count 1"));
assert!(out.contains("mytheclipse_active_threads 4"));
assert!(out.contains("# TYPE my_reqs counter"));
assert!(out.contains("my_reqs 7"));
assert!(out.contains("# TYPE my_temp gauge"));
assert!(out.contains("my_temp 1.5"));
assert!(out.contains("my_lat_ms_count 1"));
assert!(out.contains("my_lat_ms_avg 12000"));
}
}
+152
View File
@@ -0,0 +1,152 @@
//! Span and panic isolation tracking (feature `observability`).
//!
//! [`PanicTracker::install`] installs a global panic hook that logs every
//! panic through [`tracing`] inside a `mytheclipse_panic_task` span — with the
//! panic message and source location — without stopping the application.
//! [`PanicTracker::catch`] wraps a closure so a panic is captured and reported
//! as a [`PanicInfo`] instead of unwinding across a boundary.
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::sync::Arc;
/// Information about a captured panic.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PanicInfo {
/// Best-effort rendering of the panic payload.
pub message: String,
/// The source location of the panic, if available.
pub location: Option<String>,
}
#[allow(deprecated)]
type PanicHookFn = dyn Fn(&std::panic::PanicInfo<'_>) + Send + Sync + 'static;
/// A RAII guard that, while held, keeps the mytheclipse panic hook installed,
/// restoring the previously-installed hook on drop.
///
/// Returned by [`PanicTracker::install`].
#[must_use = "dropping the guard restores the previous panic hook"]
pub struct PanicGuard {
previous: Arc<PanicHookFn>,
}
/// Installs a panic hook that logs panics through [`tracing`].
pub struct PanicTracker;
impl PanicTracker {
/// Installs a global panic hook that records each panic via
/// `tracing::error!` inside a `mytheclipse_panic_task` span, then re-invokes
/// the previously-installed hook so default panic output (and any user
/// hook) still runs.
///
/// The application does not stop; the panic still unwinds normally, but a
/// trace is captured first. Returns a [`PanicGuard`] that restores the
/// previous hook when dropped.
#[allow(deprecated)]
pub fn install() -> PanicGuard {
let previous: Box<PanicHookFn> = std::panic::take_hook();
let previous: Arc<PanicHookFn> = Arc::from(previous);
let for_hook = Arc::clone(&previous);
std::panic::set_hook(Box::new(move |info| {
let span = tracing::error_span!("mytheclipse_panic_task");
let message = payload_to_string(info.payload());
let location = info
.location()
.map(|l| format!("{}:{}:{}", l.file(), l.line(), l.column()));
let _guard = span.enter();
match location {
Some(loc) => tracing::error!("panic caught: {message} at {loc}"),
None => tracing::error!("panic caught: {message}"),
}
for_hook(info);
}));
PanicGuard { previous }
}
/// Runs `f`, catching any panic and returning it as an
/// [`Err(PanicInfo)`](PanicInfo) rather than unwinding across the caller.
///
/// Useful to isolate a panic inside a worker thread so it cannot bring
/// down the rest of the process. The caller is responsible for deciding
/// whether to continue after a caught panic.
pub fn catch<T>(f: impl FnOnce() -> T) -> Result<T, PanicInfo> {
let wrapped = AssertUnwindSafe(f);
catch_unwind(wrapped).map_err(|payload| PanicInfo {
message: payload_to_string(&*payload),
location: None,
})
}
}
impl Drop for PanicGuard {
fn drop(&mut self) {
// Take out the hook we installed so we don't call ourselves, then
// restore the one that was active before it.
let _current = std::panic::take_hook();
let previous = Arc::clone(&self.previous);
std::panic::set_hook(Box::new(move |info| previous(info)));
}
}
/// Renders a panic payload (a `&str`, `String`, or fallback) to a string.
fn payload_to_string(payload: &(dyn std::any::Any + Send)) -> String {
if let Some(message) = payload.downcast_ref::<&str>() {
(*message).to_string()
} else if let Some(message) = payload.downcast_ref::<String>() {
message.clone()
} else {
"panic with a non-string payload".to_string()
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
#[test]
fn catch_returns_ok_on_success() {
let result = PanicTracker::catch(|| 1 + 1);
assert_eq!(result.unwrap(), 2);
}
#[test]
fn catch_captures_panic_as_err() {
let result: Result<u32, PanicInfo> = PanicTracker::catch(|| panic!("boom"));
assert!(result.is_err());
assert_eq!(result.unwrap_err().message, "boom");
}
#[test]
fn catch_captures_string_payload() {
let err = PanicTracker::catch(|| panic!("{}", String::from("stringy"))).unwrap_err();
assert_eq!(err.message, "stringy");
}
#[test]
fn hook_is_restored_after_guard_drop() {
// Install then drop — previous hook (default) restored.
let _guard = PanicTracker::install();
drop(_guard);
let _ = PanicTracker::catch(|| panic!("irrelevant"));
}
#[test]
fn hook_runs_without_and_panics_are_still_catchable() {
let calls = Arc::new(AtomicUsize::new(0));
let calls_hook = Arc::clone(&calls);
let prev = std::panic::take_hook();
std::panic::set_hook(Box::new(move |info| {
calls_hook.fetch_add(1, Ordering::SeqCst);
let _ = info.location().is_some();
}));
let _guard = PanicTracker::install();
// A panic in another thread triggers the hook but not the main thread.
std::thread::spawn(|| panic!("worker boom"))
.join()
.unwrap_err();
drop(_guard);
std::panic::set_hook(prev);
assert!(calls.load(Ordering::SeqCst) >= 1);
}
}
+189
View File
@@ -0,0 +1,189 @@
//! A token-bucket rate limiter (feature `traffic`).
//!
//! [`RateLimiter`] caps how many tokens can be consumed per unit time, so a
//! caller/request stream is throttled (e.g. at most 100 jobs per second).
//! Tokens are refilled lazily from elapsed time rather than by a background
//! task, and a configurable burst capacity permits short bursts beyond the
//! steady-state rate.
use std::sync::Mutex;
use std::time::{Duration, Instant};
use tracing::Instrument;
use crate::error::MytheclipseError;
/// The error returned by rate-limit acquisition.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RateLimitError {
/// The limiter's capacity was exceeded and no retry path was taken.
///
/// Returned by [`RateLimiter::try_acquire`] when no token is available, or
/// wrapped into the shared [`MytheclipseError::RateLimited`] variant.
Limited,
}
impl std::fmt::Display for RateLimitError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Limited => write!(f, "rate limit exceeded"),
}
}
}
impl std::error::Error for RateLimitError {}
struct Inner {
capacity: u64,
refill_per_sec: f64,
tokens: Mutex<f64>,
last_refill: Mutex<Instant>,
}
/// A thread-safe, async-compatible token-bucket rate limiter.
///
/// Construct with [`RateLimiter::new`], then [`RateLimiter::acquire`] to await
/// a token or [`RateLimiter::try_acquire`] to fail fast when exhausted.
#[derive(Clone)]
pub struct RateLimiter {
inner: std::sync::Arc<Inner>,
}
impl RateLimiter {
/// Builds a limiter refilling at `rate_per_sec` tokens/second with a burst
/// capacity of `burst_capacity`.
///
/// # Panics
///
/// Panics if `rate_per_sec` is not finite and positive, or if
/// `burst_capacity` is zero.
pub fn new(rate_per_sec: f64, burst_capacity: u64) -> Self {
assert!(
rate_per_sec.is_finite() && rate_per_sec > 0.0,
"rate must be positive"
);
assert!(burst_capacity > 0, "burst capacity must be > 0");
Self {
inner: std::sync::Arc::new(Inner {
capacity: burst_capacity,
refill_per_sec: rate_per_sec,
tokens: Mutex::new(burst_capacity as f64),
last_refill: Mutex::new(Instant::now()),
}),
}
}
/// Awaits until a token is available and consumes one.
///
/// Returns [`RateLimitError::Limited`] (in practice, never for `acquire`)
/// on closure; the suspended waiter is woken when the bucket refills. The
/// wait is wrapped in a `mytheclipse_ratelimit_task` tracing span.
pub async fn acquire(&self) -> Result<(), RateLimitError> {
loop {
if self.try_acquire().is_ok() {
return Ok(());
}
let span = tracing::info_span!("mytheclipse_ratelimit_task");
// Sleep for the time it takes to refill a single token.
let sleep = Duration::from_secs_f64(1.0 / self.inner.refill_per_sec);
tokio::time::sleep(sleep).instrument(span).await;
}
}
/// Consumes one token immediately if one is available.
///
/// Returns `Err(RateLimitError::Limited)` (or the caller may map it to
/// [`MytheclipseError::RateLimited`]) when the bucket is empty.
pub fn try_acquire(&self) -> Result<(), RateLimitError> {
let mut tokens = self.inner.tokens.lock().unwrap();
let mut last = self.inner.last_refill.lock().unwrap();
self.refill(&mut tokens, *last);
*last = Instant::now();
if *tokens >= 1.0 {
*tokens -= 1.0;
Ok(())
} else {
Err(RateLimitError::Limited)
}
}
/// The approximate number of tokens currently available (including burst
/// headroom), for metrics/observability purposes.
pub fn available_tokens(&self) -> u64 {
let mut tokens = self.inner.tokens.lock().unwrap();
let last = self.inner.last_refill.lock().unwrap();
self.refill(&mut tokens, *last);
tokens.floor() as u64
}
fn refill(&self, tokens: &mut f64, last: Instant) {
let elapsed_secs = last.elapsed().as_secs_f64();
let added = elapsed_secs * self.inner.refill_per_sec;
*tokens = (*tokens + added).min(self.inner.capacity as f64);
}
/// `acquire` mapped onto the shared error type, e.g. for code returning
/// [`MytheclipseError`].
pub async fn acquire_err(&self) -> Result<(), MytheclipseError> {
self.acquire()
.await
.map_err(|_| MytheclipseError::RateLimited)
}
/// `try_acquire` mapped onto the shared error type.
pub fn try_acquire_err(&self) -> Result<(), MytheclipseError> {
self.try_acquire()
.map_err(|_| MytheclipseError::RateLimited)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn burst_capacity_consumed_immediately() {
let limiter = RateLimiter::new(1000.0, 3);
assert!(limiter.try_acquire().is_ok());
assert!(limiter.try_acquire().is_ok());
assert!(limiter.try_acquire().is_ok());
assert_eq!(limiter.try_acquire(), Err(RateLimitError::Limited));
}
#[test]
fn available_tokens_bounded_by_capacity() {
let limiter = RateLimiter::new(1000.0, 5);
assert_eq!(limiter.available_tokens(), 5);
let _ = limiter.try_acquire();
assert_eq!(limiter.available_tokens(), 4);
}
#[tokio::test]
async fn acquire_waits_for_refill() {
let limiter = RateLimiter::new(1000.0, 1);
assert!(limiter.try_acquire().is_ok());
let start = Instant::now();
limiter
.acquire()
.await
.expect("acquire should eventually succeed");
assert!(start.elapsed() >= Duration::from_millis(1));
}
#[tokio::test]
async fn try_acquire_err_maps_to_shared_error() {
let limiter = RateLimiter::new(1000.0, 1);
let _ = limiter.try_acquire();
assert_eq!(
limiter.try_acquire_err(),
Err(MytheclipseError::RateLimited)
);
}
#[test]
#[should_panic]
fn zero_burst_panics() {
let _ = RateLimiter::new(1.0, 0);
}
}
+256
View File
@@ -0,0 +1,256 @@
//! Automatic retry with exponential backoff and jitter (feature `resiliency`).
//!
//! [`retry`] re-runs a fallible async operation according to a
//! [`RetryConfig`], sleeping an exponentially-growing, optionally-jittered
//! delay between attempts. A predicate decides which errors are retryable, so
//! permanent failures (e.g. a 4xx response) short-circuit immediately while
//! transient ones (network hiccups, connection refused) are retried.
use std::future::Future;
use std::time::Duration;
use rand::Rng;
use tracing::Instrument;
/// How much random jitter to apply to each backoff delay.
///
/// Jitter prevents the "thundering herd" of many retrying clients waking
/// simultaneously; [`Full`](JitterKind::Full) is the most aggressive and is
/// the recommended default for distributed systems.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JitterKind {
/// No randomness: sleep exactly the computed backoff.
None,
/// Sleep in `[delay, 2 * delay)` — sometimes called "equal jitter".
Equal,
/// Sleep in `[0, delay)` — "full jitter".
Full,
}
/// Configuration driving [`retry`].
#[derive(Debug, Clone)]
pub struct RetryConfig {
/// Total number of attempts (including the first). Must be `>= 1`.
pub max_attempts: u32,
/// Initial delay before the first retry.
pub base_delay: Duration,
/// Upper bound on the computed backoff delay.
pub max_delay: Duration,
/// Exponential growth factor applied after each failure.
pub factor: f64,
/// Jitter strategy applied to each delay.
pub jitter: JitterKind,
}
impl Default for RetryConfig {
fn default() -> Self {
Self {
max_attempts: 3,
base_delay: Duration::from_millis(500),
max_delay: Duration::from_secs(10),
factor: 2.0,
jitter: JitterKind::Full,
}
}
}
/// The error returned by [`retry`] once it gives up.
#[derive(Debug)]
pub enum RetryError<E> {
/// All attempts were made and the last one failed.
///
/// `attempts` is the total number of attempts performed and `last` is the
/// error produced by the final attempt.
Exhausted { attempts: u32, last: E },
}
impl<E> RetryError<E> {
/// Returns a reference to the error produced by the final attempt.
pub fn last(&self) -> &E {
match self {
Self::Exhausted { last, .. } => last,
}
}
}
impl<E: std::fmt::Display> std::fmt::Display for RetryError<E> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Exhausted { attempts, last } => {
write!(f, "retry exhausted after {attempts} attempts: {last}")
}
}
}
}
impl<E: std::fmt::Debug + std::fmt::Display> std::error::Error for RetryError<E> {}
/// Retries `op` according to `config`, retrying only errors for which
/// `filter` returns `true`.
///
/// On the first attempt and after each retryable failure, `op` is invoked
/// again (rebuilt from the captured closure) until it succeeds, the retryable
/// error becomes permanent, or `max_attempts` is reached. Between attempts the
/// coroutine sleeps for a backoff delay computed as
/// `min(max_delay, base_delay * factor^attempt)` with jitter applied per
/// [`RetryConfig::jitter`].
///
/// Each attempt runs inside a `mytheclipse_retry_task` tracing span carrying
/// the attempt index and total.
pub async fn retry<T, E, F, Fut, P>(
config: RetryConfig,
filter: P,
mut op: F,
) -> Result<T, RetryError<E>>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<T, E>>,
P: Fn(&E) -> bool,
{
let mut attempt: u32 = 0;
loop {
attempt += 1;
let span = tracing::info_span!(
"mytheclipse_retry_task",
attempt,
max_attempts = config.max_attempts
);
let result = op().instrument(span).await;
match result {
Ok(value) => return Ok(value),
Err(err) => {
let retryable = filter(&err);
if !retryable || attempt >= config.max_attempts {
return Err(RetryError::Exhausted {
attempts: attempt,
last: err,
});
}
let delay = backoff_delay(&config, attempt, rand::thread_rng());
tokio::time::sleep(delay).await;
}
}
}
}
/// Computes the (jittered) delay to sleep before retry `attempt` (1-based).
///
/// Kept as a pure function for testability.
fn backoff_delay<R: Rng>(config: &RetryConfig, attempt: u32, mut rng: R) -> Duration {
let exponent = attempt.saturating_sub(1) as f64; // first retry uses base
let computed = config.base_delay.as_millis() as f64 * config.factor.powf(exponent);
let max_ms = config.max_delay.as_millis() as f64;
let capped = computed.min(max_ms);
let millis = match config.jitter {
JitterKind::None => capped,
JitterKind::Equal => capped / 2.0 + rng.gen_range(0.0..capped / 2.0),
JitterKind::Full => rng.gen_range(0.0..capped),
};
Duration::from_millis(millis.clamp(0.0, max_ms) as u64)
}
#[cfg(test)]
mod tests {
use super::*;
use std::cell::Cell;
fn always<E>(_: &E) -> bool {
true
}
#[tokio::test]
async fn succeeds_on_first_attempt() {
let calls = Cell::new(0u32);
let result = retry(RetryConfig::default(), always, || async {
calls.set(calls.get() + 1);
Ok::<_, ()>(42u32)
})
.await;
assert_eq!(result.unwrap(), 42);
assert_eq!(calls.get(), 1);
}
#[tokio::test]
async fn succeeds_after_transient_failures() {
let config = RetryConfig {
max_attempts: 5,
base_delay: Duration::from_millis(1),
..RetryConfig::default()
};
let calls = Cell::new(0u32);
let result = retry(config, always, || async {
calls.set(calls.get() + 1);
if calls.get() < 3 {
Err::<u32, u8>(9)
} else {
Ok(7u32)
}
})
.await;
assert_eq!(result.unwrap(), 7);
assert_eq!(calls.get(), 3);
}
#[tokio::test]
async fn exhausts_after_max_attempts() {
let config = RetryConfig {
max_attempts: 3,
base_delay: Duration::from_millis(1),
..RetryConfig::default()
};
let calls = Cell::new(0u32);
let result = retry(config, always, || async {
calls.set(calls.get() + 1);
Err::<u32, u8>(42)
})
.await;
assert!(matches!(
result,
Err(RetryError::Exhausted { attempts: 3, .. })
));
assert_eq!(result.unwrap_err().last(), &42);
}
#[tokio::test]
async fn non_retryable_error_short_circuits() {
let config = RetryConfig {
max_attempts: 10,
base_delay: Duration::from_millis(1),
..RetryConfig::default()
};
let calls = Cell::new(0u32);
let result = retry(
config,
|e: &u16| *e != 403,
|| async {
calls.set(calls.get() + 1);
Err::<u32, u16>(403)
},
)
.await;
assert!(matches!(
result,
Err(RetryError::Exhausted { attempts: 1, .. })
));
assert_eq!(calls.get(), 1);
}
#[test]
fn full_jitter_is_within_bounds_and_capped() {
let config = RetryConfig {
max_attempts: 3,
base_delay: Duration::from_secs(2),
max_delay: Duration::from_secs(4),
factor: 10.0,
jitter: JitterKind::Full,
};
let mut rng = rand::thread_rng();
for _ in 0..1000 {
let d = backoff_delay(&config, 2, &mut rng);
assert!(d <= config.max_delay);
}
}
}
+214
View File
@@ -0,0 +1,214 @@
//! Graceful shutdown coordination (feature `lifecycle`).
//!
//! [`ShutdownManager`] watches for OS signals (SIGINT/SIGTERM on Unix, Ctrl-C
//! everywhere) and broadcasts a shutdown signal to every registered task, then
//! waits for those tasks to finish before the process exits. Tasks subscribe
//! via a cloneable [`ShutdownSignal`] and cooperatively stop when it fires.
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::watch;
use tracing::Instrument;
/// A cloneable token a task holds to observe whether shutdown has been
/// requested.
#[derive(Clone)]
pub struct ShutdownSignal {
rx: watch::Receiver<bool>,
}
impl ShutdownSignal {
/// Whether shutdown has been requested.
pub fn is_shutdown(&self) -> bool {
*self.rx.borrow()
}
/// Awaits until shutdown is requested, then returns.
pub async fn wait(&mut self) {
if *self.rx.borrow() {
return;
}
// Borrow avoids holding the receiver while yielding.
let _ = self.rx.changed().await;
}
}
/// Coordinates graceful shutdown of tracked background tasks.
///
/// Construct with [`ShutdownManager::new`], hand each long-running task a
/// [`ShutdownSignal`] via [`ShutdownManager::handle`], register its join
/// handle with [`ShutdownManager::register`], and finally defer the process
/// exit until [`ShutdownManager::drain`] completes.
#[derive(Clone)]
pub struct ShutdownManager {
inner: Arc<Inner>,
}
struct Inner {
tx: watch::Sender<bool>,
/// Kept alive so `tx.send` always has a receiver and therefore always
/// updates the stored value — even if `request()` fires before any
/// [`ShutdownSignal`] has been handed out.
_keepalive: watch::Receiver<bool>,
tasks: std::sync::Mutex<Vec<tokio::task::JoinHandle<()>>>,
}
impl Default for ShutdownManager {
fn default() -> Self {
Self::new()
}
}
impl ShutdownManager {
/// Builds a new shutdown manager with no tasks tracked.
pub fn new() -> Self {
let (tx, keepalive) = watch::channel(false);
Self {
inner: Arc::new(Inner {
tx,
_keepalive: keepalive,
tasks: std::sync::Mutex::new(Vec::new()),
}),
}
}
/// Returns a new [`ShutdownSignal`] this manager will fire on shutdown.
pub fn handle(&self) -> ShutdownSignal {
ShutdownSignal {
rx: self.inner.tx.subscribe(),
}
}
/// Registers a background task so shutdown waits for it to complete.
///
/// The task is expected to observe its [`ShutdownSignal`] and stop
/// promptly once it fires; `drain` gives tasks a grace window.
pub fn register(&self, handle: tokio::task::JoinHandle<()>) {
self.inner.tasks.lock().unwrap().push(handle);
}
/// Whether shutdown has been requested.
pub fn is_shutdown(&self) -> bool {
*self.inner.tx.borrow()
}
/// Awaits until shutdown is requested (by an OS signal or an explicit
/// [`ShutdownManager::request`]).
pub async fn wait_for_shutdown(&self) {
let mut rx = self.inner.tx.subscribe();
if *rx.borrow() {
return;
}
let _ = rx.changed().await;
}
/// Requests shutdown programmatically (also invoked by the signal
/// handler). Safe to call more than once.
pub fn request(&self) {
let _ = self.inner.tx.send(true);
}
/// Waits for shutdown to be requested, then awaits all registered tasks,
/// allowing at most `grace` per task before giving up.
///
/// Runs in a `mytheclipse_shutdown_task` tracing span.
pub async fn drain(&self, grace: Duration) {
let span = tracing::info_span!("mytheclipse_shutdown_task");
self.wait_for_shutdown().instrument(span.clone()).await;
let tasks = {
let mut guard = self.inner.tasks.lock().unwrap();
std::mem::take(&mut *guard)
};
for task in tasks {
let _ = tokio::time::timeout(grace, task)
.instrument(span.clone())
.await;
}
}
/// Waits for the next OS termination signal (SIGINT/SIGTERM on Unix,
/// Ctrl-C elsewhere).
pub async fn wait_for_os_signal(&self) {
let _ = os_signal().await;
}
}
/// Awaits SIGINT/SIGTERM on Unix, or Ctrl-C on other platforms.
#[cfg(unix)]
async fn os_signal() {
use tokio::signal::unix::{signal, SignalKind};
let mut sigint = signal(SignalKind::interrupt()).expect("failed to install SIGINT handler");
let mut sigterm = signal(SignalKind::terminate()).expect("failed to install SIGTERM handler");
tokio::select! {
_ = sigint.recv() => {},
_ = sigterm.recv() => {},
}
}
/// Awaits Ctrl-C on non-Unix platforms.
#[cfg(not(unix))]
async fn os_signal() {
use tokio::signal::ctrl_c;
let _ = ctrl_c().await;
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn signal_fires_on_request() {
let manager = ShutdownManager::new();
let mut sig = manager.handle();
assert!(!sig.is_shutdown());
assert!(!manager.is_shutdown());
let observer = tokio::spawn(async move {
sig.wait().await;
sig.is_shutdown()
});
manager.request();
assert!(manager.is_shutdown());
assert!(observer.await.unwrap());
}
#[tokio::test]
async fn request_is_idempotent() {
let manager = ShutdownManager::new();
manager.request();
manager.request();
let sig = manager.handle();
assert!(sig.is_shutdown());
}
#[tokio::test]
async fn drain_waits_for_registered_tasks() {
let manager = ShutdownManager::new();
let sig = manager.handle();
let handle = tokio::spawn(async move {
let mut sig = sig;
sig.wait().await;
});
manager.register(handle);
manager.request();
// drain completes promptly because the task stops on the signal.
manager.drain(Duration::from_secs(5)).await;
}
#[tokio::test]
async fn drain_times_out_a_slow_task() {
let manager = ShutdownManager::new();
let _sig = manager.handle();
// A task that never observes shutdown — it must be timed out.
let slow = tokio::spawn(std::future::pending::<()>());
manager.register(slow);
manager.request();
// drain should return (time out) rather than hang forever.
manager.drain(Duration::from_millis(50)).await;
}
}
+155
View File
@@ -0,0 +1,155 @@
//! Hard execution time bounds for async work (feature `resiliency`).
//!
//! Provides [`with_timeout`] (a convenience that resolves a future to a
//! [`Result`] with an elapsed-vs-completed outcome) and [`Timeout`], a
//! stand-alone [`Future`] wrapper you can build once and hand to
//! [`crate::io::spawn_io`] / [`crate::bg::spawn_bg`]/`tokio::spawn` so the
//! bound is enforced wherever the future actually runs.
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use tracing::Instrument;
use crate::error::MytheclipseError;
/// The outcome of timing out a single future.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TimeoutError {
/// The deadline elapsed before the future completed.
Elapsed,
}
impl std::fmt::Display for TimeoutError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Elapsed => write!(f, "deadline elapsed"),
}
}
}
impl std::error::Error for TimeoutError {}
/// Runs `future` to completion, giving up after `dur` and returning
/// [`TimeoutError::Elapsed`] if the bound is exceeded.
///
/// The future is cancelled on timeout: its task is aborted when the deadline
/// fires, so no lingering resource is pinned. The call is wrapped in a
/// `mytheclipse_timeout_task` tracing span.
///
/// # Panics
///
/// Panics if called outside the context of a running Tokio runtime.
pub async fn with_timeout<T, F>(dur: Duration, future: F) -> Result<T, TimeoutError>
where
F: Future<Output = T>,
{
let span = tracing::info_span!("mytheclipse_timeout_task");
match tokio::time::timeout(dur, future.instrument(span)).await {
Ok(value) => Ok(value),
Err(_) => Err(TimeoutError::Elapsed),
}
}
/// A future that enforces a [`Duration`] deadline on an inner future.
///
/// Unlike [`with_timeout`], this returns an invocable [`Future`] rather than
/// polling to completion, so it can be constructed ahead of time and spawned
/// through any executor:
///
/// ```no_run
/// use mytheclipse::timeout::Timeout;
/// # fn _r() {
/// let bounded: Timeout<_> = Timeout::new(
/// std::time::Duration::from_secs(1),
/// async { "done" },
/// );
/// # }
/// ```
///
/// If the deadline elapses first, the completed output is `Err(TimeoutError)`.
pub struct Timeout<T> {
inner: Pin<Box<dyn Future<Output = Result<T, TimeoutError>> + Send>>,
}
impl<T: Send + 'static> Timeout<T> {
/// Wraps `future` with a deadline of `dur`.
pub fn new<F>(dur: Duration, future: F) -> Self
where
F: Future<Output = T> + Send + 'static,
{
let future = async move {
match tokio::time::timeout(dur, future).await {
Ok(value) => Ok(value),
Err(_) => Err(TimeoutError::Elapsed),
}
};
Self {
inner: Box::pin(future),
}
}
}
impl<T> Future for Timeout<T> {
type Output = Result<T, TimeoutError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// `inner` is a `Pin<Box<...>>`. Because `Box` is `Unpin`, calling
// `as_mut` on the boxed pin yields `Pin<&mut (dyn Future + Send)>`
// which polls the inner future. Pinning is upheld by the box, so no
// unsafe projection is needed.
self.get_mut().inner.as_mut().poll(cx)
}
}
/// A hard deadline for a single future, returning `Result<T, MytheclipseError>`.
///
/// Convenience twin of [`with_timeout`] that maps the outcome onto the crate's
/// shared error type, e.g. for use in code that already returns
/// [`MytheclipseError`].
pub async fn timeout<T, F>(dur: Duration, future: F) -> Result<T, MytheclipseError>
where
F: Future<Output = T>,
{
with_timeout(dur, future)
.await
.map_err(|_| MytheclipseError::Timeout)
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn completes_within_bound_returns_value() {
let value = with_timeout(Duration::from_secs(5), async { 42u32 }).await;
assert_eq!(value.unwrap(), 42);
}
#[tokio::test]
async fn exceeding_bound_yields_elapsed() {
let outcome = with_timeout(Duration::from_millis(20), async {
tokio::time::sleep(Duration::from_secs(5)).await;
42u32
})
.await;
assert_eq!(outcome, Err(TimeoutError::Elapsed));
}
#[tokio::test]
async fn timeout_wrapper_maps_to_shared_error() {
let outcome = timeout(Duration::from_millis(10), async {
tokio::time::sleep(Duration::from_secs(5)).await;
})
.await;
assert_eq!(outcome, Err(MytheclipseError::Timeout));
}
#[tokio::test]
async fn timeout_future_is_spawnable() {
let bounded = Timeout::new(Duration::from_secs(5), async { 7u32 });
assert_eq!(tokio::spawn(bounded).await.unwrap().unwrap(), 7);
}
}