diff --git a/crates/taskito-core/src/resilience/circuit_breaker.rs b/crates/taskito-core/src/resilience/circuit_breaker.rs index c85d7ea4..f60bdd99 100644 --- a/crates/taskito-core/src/resilience/circuit_breaker.rs +++ b/crates/taskito-core/src/resilience/circuit_breaker.rs @@ -50,20 +50,22 @@ pub struct CircuitBreakerConfig { pub half_open_success_rate: f64, } -/// Circuit breaker manager backed by SQLite. -pub struct CircuitBreaker { +/// Circuit breaker manager backed by SQLite. In-crate only: consumed solely by +/// `Scheduler`; the config/state contract is exposed via `CircuitBreakerConfig` +/// and `CircuitState`. +pub(crate) struct CircuitBreaker { storage: StorageBackend, } impl CircuitBreaker { /// Build a breaker manager over `storage`. - pub fn new(storage: StorageBackend) -> Self { + pub(crate) fn new(storage: StorageBackend) -> Self { Self { storage } } /// Check if a task is allowed to execute. Returns true if allowed. /// Transitions Open -> HalfOpen after cooldown. - pub fn allow(&self, task_name: &str) -> Result { + pub(crate) fn allow(&self, task_name: &str) -> Result { let row = match self.storage.get_circuit_breaker(task_name)? { Some(r) => r, None => return Ok(true), // No breaker configured = always allow @@ -125,7 +127,7 @@ impl CircuitBreaker { /// Record a task success. In HalfOpen, tracks probes and closes when /// the success rate threshold is met. - pub fn record_success(&self, task_name: &str) -> Result<()> { + pub(crate) fn record_success(&self, task_name: &str) -> Result<()> { let row = match self.storage.get_circuit_breaker(task_name)? { Some(r) => r, None => return Ok(()), @@ -197,7 +199,7 @@ impl CircuitBreaker { } /// Record a task failure. May trip the breaker open. - pub fn record_failure(&self, task_name: &str) -> Result<()> { + pub(crate) fn record_failure(&self, task_name: &str) -> Result<()> { let now = now_millis(); let row = match self.storage.get_circuit_breaker(task_name)? { @@ -295,7 +297,7 @@ impl CircuitBreaker { } /// Register a circuit breaker for a task (idempotent). - pub fn register(&self, task_name: &str, config: &CircuitBreakerConfig) -> Result<()> { + pub(crate) fn register(&self, task_name: &str, config: &CircuitBreakerConfig) -> Result<()> { if self.storage.get_circuit_breaker(task_name)?.is_some() { return Ok(()); } diff --git a/crates/taskito-core/src/resilience/dlq.rs b/crates/taskito-core/src/resilience/dlq.rs index ebe2c483..15dcba9b 100644 --- a/crates/taskito-core/src/resilience/dlq.rs +++ b/crates/taskito-core/src/resilience/dlq.rs @@ -1,35 +1,21 @@ use crate::error::Result; use crate::job::Job; -use crate::storage::{DeadJob, Storage, StorageBackend}; +use crate::storage::{Storage, StorageBackend}; -/// Dead letter queue manager. -pub struct DeadLetterQueue { +/// Dead letter queue manager. In-crate only: consumed solely by `Scheduler`; +/// bindings drive the DLQ through `Storage` methods directly. +pub(crate) struct DeadLetterQueue { storage: StorageBackend, } impl DeadLetterQueue { /// Build a DLQ manager over `storage`. - pub fn new(storage: StorageBackend) -> Self { + pub(crate) fn new(storage: StorageBackend) -> Self { Self { storage } } /// Move a failed job to the dead letter queue. - pub fn move_to_dlq(&self, job: &Job, error: &str, metadata: Option<&str>) -> Result<()> { + pub(crate) fn move_to_dlq(&self, job: &Job, error: &str, metadata: Option<&str>) -> Result<()> { self.storage.move_to_dlq(job, error, metadata) } - - /// List dead letter entries. - pub fn list(&self, limit: i64, offset: i64) -> Result> { - self.storage.list_dead(limit, offset) - } - - /// Re-enqueue a dead letter job. Returns the new job ID. - pub fn retry(&self, dead_id: &str) -> Result { - self.storage.retry_dead(dead_id) - } - - /// Purge dead letter entries older than the given number of milliseconds ago. - pub fn purge(&self, older_than_ms: i64) -> Result { - self.storage.purge_dead(older_than_ms) - } } diff --git a/crates/taskito-core/src/resilience/rate_limiter.rs b/crates/taskito-core/src/resilience/rate_limiter.rs index a5c7bf6d..6adeecbf 100644 --- a/crates/taskito-core/src/resilience/rate_limiter.rs +++ b/crates/taskito-core/src/resilience/rate_limiter.rs @@ -1,8 +1,9 @@ use crate::error::Result; use crate::storage::{Storage, StorageBackend}; -/// Token bucket rate limiter backed by SQLite for persistence. -pub struct RateLimiter { +/// Token bucket rate limiter backed by SQLite for persistence. In-crate only: +/// consumed solely by `Scheduler`; the config contract is `RateLimitConfig`. +pub(crate) struct RateLimiter { storage: StorageBackend, } @@ -40,14 +41,14 @@ impl RateLimitConfig { impl RateLimiter { /// Build a rate limiter over `storage`. - pub fn new(storage: StorageBackend) -> Self { + pub(crate) fn new(storage: StorageBackend) -> Self { Self { storage } } /// Try to acquire a token for the given key. /// Returns `true` if the token was acquired, `false` if rate limited. /// Uses an atomic transaction to prevent race conditions. - pub fn try_acquire(&self, key: &str, config: &RateLimitConfig) -> Result { + pub(crate) fn try_acquire(&self, key: &str, config: &RateLimitConfig) -> Result { self.storage .try_acquire_token(key, config.max_tokens, config.refill_rate) } diff --git a/crates/taskito-core/src/scheduler/mod.rs b/crates/taskito-core/src/scheduler/mod.rs index 3a979fcb..cc5c54bd 100644 --- a/crates/taskito-core/src/scheduler/mod.rs +++ b/crates/taskito-core/src/scheduler/mod.rs @@ -575,11 +575,6 @@ impl Scheduler { self.task_configs.insert(task_name, config); } - /// The circuit-breaker manager shared with this scheduler's storage. - pub fn circuit_breaker(&self) -> &CircuitBreaker { - &self.circuit_breaker - } - /// Run the scheduler loop. Polls for ready jobs and dispatches them /// to the worker pool via the provided channel. ///