diff --git a/CHANGELOG.md b/CHANGELOG.md index d8bd4de1..79086fa4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +### Added + +- Introduced a Redis-backed distributed scheduler backend with configurable lock + management and task state persistence, plus documentation and examples for + running multi-instance beat deployments. + ### Changed - Updated GitHub Actions to latest versions to resolve deprecation warnings diff --git a/Cargo.toml b/Cargo.toml index 0321a9ef..d4e52171 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,6 +23,9 @@ path = "src/lib.rs" [[example]] name = "celery_app" +[[example]] +name = "redis_beat" + [dependencies] base64 = "0.22.1" chrono = { version = "0.4.42", features = ["serde"] } @@ -46,7 +49,7 @@ colored = "3.0.0" once_cell = { version = "1.21.3" } globset = "0.4.16" hostname = "0.4.1" -redis = { version = "0.32.5", features=["connection-manager", "tokio-comp"] } +redis = { version = "0.32.5", features=["connection-manager", "tokio-comp", "script"] } deadpool-redis = { version = "0.22.0", features = ["rt_tokio_1"] } tokio-executor-trait = "3.1.0" tokio-reactor-trait = "2.8.0" diff --git a/README.md b/README.md index a4a31773..19587a85 100644 --- a/README.md +++ b/README.md @@ -140,6 +140,26 @@ cargo run --example beat_app And then you can consume tasks from Rust or Python as explained above. +#### Redis-backed Beat + +A Redis-powered distributed scheduler backend is available through `RedisSchedulerBackend`. +To try it out locally (requires a Redis server running): + +```bash +REDIS_URL=redis://127.0.0.1:6379/0 cargo run --example redis_beat +``` + +Only the instance that holds the Redis lock will dispatch tasks, while followers wait and +take over automatically if the leader disconnects. + +To test multi-instance failover: + +1. Run a worker connected to Redis so scheduled tasks are consumed (see `examples/celery_app.rs`). +2. Start the first beat instance as shown above; it will log that it acquired leadership. +3. Start a second beat instance with the same command; it stays on standby. +4. Stop the first instance (e.g. Ctrl+C). Within a few seconds the standby will acquire the lock + and resume scheduling without losing any tasks. + ## Road map and current state ✅ = Supported and mostly stable, although there may be a few incomplete features.
@@ -148,31 +168,31 @@ And then you can consume tasks from Rust or Python as explained above. ### Core -> **Note**: Issue tracking links below reference the original repository where development history is maintained. +> **Note**: Issue tracking links below reference this repository. | | Status | Tracking | | ---------------- |:-------:| --------- | -| Protocol | ⚠️ | [![](https://img.shields.io/github/issues/rusty-celery/rusty-celery/Protocol%20Feature?label=Issues)](https://github.com/rusty-celery/rusty-celery/issues?q=is%3Aissue+label%3A%22Protocol+Feature%22+is%3Aopen) | +| Protocol | ⚠️ | [Open issues](https://github.com/GaiaNet-AI/celery-rs/issues?q=is%3Aissue+label%3A%22Protocol%20Feature%22+is%3Aopen) | | Producers | ✅ | | | Consumers | ✅ | | | Brokers | ✅ | | | Beat | ✅ | | -| Backends | 🔴 | | -| [Baskets](https://github.com/rusty-celery/rusty-celery/issues/53) | 🔴 | | +| Backends | ⚠️ | | +| Baskets | 🔴 | | ### Brokers | | Status | Tracking | | ----- |:------:| -------- | -| AMQP | ✅ | [![](https://img.shields.io/github/issues/rusty-celery/rusty-celery/Broker%3A%20AMQP?label=Issues)](https://github.com/rusty-celery/rusty-celery/labels/Broker%3A%20AMQP) | -| Redis | ✅ | [![](https://img.shields.io/github/issues/rusty-celery/rusty-celery/Broker%3A%20Redis?label=Issues)](https://github.com/rusty-celery/rusty-celery/labels/Broker%3A%20Redis) | +| AMQP | ✅ | [Open issues](https://github.com/GaiaNet-AI/celery-rs/issues?q=is%3Aissue+label%3A%22Broker%3A%20AMQP%22+is%3Aopen) | +| Redis | ✅ | [Open issues](https://github.com/GaiaNet-AI/celery-rs/issues?q=is%3Aissue+label%3A%22Broker%3A%20Redis%22+is%3Aopen) | ### Backends | | Status | Tracking | | ----------- |:------:| -------- | -| RPC | 🔴 | [![](https://img.shields.io/github/issues/rusty-celery/rusty-celery/Backend%3A%20RPC?label=Issues)](https://github.com/rusty-celery/rusty-celery/labels/Backend%3A%20RPC) | -| Redis | 🔴 | [![](https://img.shields.io/github/issues/rusty-celery/rusty-celery/Backend%3A%20Redis?label=Issues)](https://github.com/rusty-celery/rusty-celery/labels/Backend%3A%20Redis) | +| RPC | 🔴 | [Open issues](https://github.com/GaiaNet-AI/celery-rs/issues?q=is%3Aissue+label%3A%22Backend%3A%20RPC%22+is%3Aopen) | +| Redis | ✅ | [Open issues](https://github.com/GaiaNet-AI/celery-rs/issues?q=is%3Aissue+label%3A%22Backend%3A%20Redis%22+is%3Aopen) | ## Project History and Maintenance diff --git a/examples/celery_app.rs b/examples/celery_app.rs index 2c364b33..de9b71d3 100644 --- a/examples/celery_app.rs +++ b/examples/celery_app.rs @@ -62,8 +62,8 @@ async fn main() -> Result<()> { let opt = CeleryOpt::from_args(); let my_app = celery::app!( - // broker = RedisBroker { std::env::var("REDIS_ADDR").unwrap_or_else(|_| "redis://127.0.0.1:6379/".into()) }, - broker = AMQPBroker { std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672".into()) }, + broker = RedisBroker { std::env::var("REDIS_ADDR").unwrap_or_else(|_| "redis://127.0.0.1:6379/".into()) }, + // broker = AMQPBroker { std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672".into()) }, tasks = [ add, buggy_task, diff --git a/examples/redis_beat.rs b/examples/redis_beat.rs new file mode 100644 index 00000000..b43e2b4e --- /dev/null +++ b/examples/redis_beat.rs @@ -0,0 +1,46 @@ +use anyhow::Result; +use celery::beat::{CronSchedule, RedisBackendConfig, RedisSchedulerBackend}; +use celery::prelude::*; +use std::time::Duration; + +#[celery::task] +fn add(x: i32, y: i32) -> TaskResult { + Ok(x + y) +} + +#[tokio::main] +async fn main() -> Result<()> { + env_logger::init(); + + let redis_url = + std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379/0".to_string()); + + let scheduler_backend = RedisSchedulerBackend::new( + RedisBackendConfig::new(&redis_url).key_prefix("celery_rs_example"), + )?; + + let mut beat = celery::beat!( + broker = RedisBroker { redis_url }, + scheduler_backend = RedisSchedulerBackend { scheduler_backend }, + tasks = [], + task_routes = ["*" => "celery"], + ) + .await?; + + let signature = add::new(1, 2).with_queue("celery"); + beat.schedule_named_task( + "add_every_minute".to_string(), + signature, + CronSchedule::from_string("*/1 * * * *")?, + ); + + let fast_signature = add::new(3, 4).with_queue("celery"); + beat.schedule_named_task( + "add_interval".to_string(), + fast_signature, + celery::beat::DeltaSchedule::new(Duration::from_secs(30)), + ); + + beat.start().await?; + Ok(()) +} diff --git a/pyproject.toml b/pyproject.toml index 41b94430..05212e9e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,3 +1,12 @@ +[project] +name = "celery-rs-python-examples" +version = "0.1.0" +requires-python = ">=3.10" +dependencies = [ + "celery>=5.3", + "redis>=4.5", +] + [tool.black] line-length = 100 diff --git a/src/beat/backend.rs b/src/beat/backend.rs index 82975bb3..ec0c9bb8 100644 --- a/src/beat/backend.rs +++ b/src/beat/backend.rs @@ -1,7 +1,10 @@ /// This module contains the definition of application-provided scheduler backends. use super::scheduled_task::ScheduledTask; use crate::error::BeatError; -use std::collections::BinaryHeap; +use std::{collections::BinaryHeap, future::Future, pin::Pin, time::Duration}; + +mod redis; +pub use redis::{RedisBackendConfig, RedisSchedulerBackend}; /// A `SchedulerBackend` is in charge of keeping track of the internal state of the scheduler /// according to some source of truth, such as a database. @@ -24,11 +27,63 @@ pub trait SchedulerBackend { /// This method will not be called if `should_sync` returns `false`. fn sync(&mut self, scheduled_tasks: &mut BinaryHeap) -> Result<(), BeatError>; + /// Return a mutable reference to the distributed capability of this backend, if any. + /// + /// Backends that do not support distributed coordination can leave the default + /// implementation untouched, and the beat loop will fall back to single-instance + /// behaviour. + fn as_distributed(&mut self) -> Option<&mut dyn DistributedScheduler> { + None + } + // Maybe we should consider some methods to inform the backend that a task has been executed. // Not sure about what Python does, but at least it keeps a counter with the number of executed tasks, // and the backend has access to that. } +pub struct TickDecision { + pub execute_tasks: bool, + pub sleep_hint: Option, +} + +impl TickDecision { + pub fn execute() -> Self { + TickDecision { + execute_tasks: true, + sleep_hint: None, + } + } + + pub fn execute_with_hint(sleep_hint: Duration) -> Self { + TickDecision { + execute_tasks: true, + sleep_hint: Some(sleep_hint), + } + } + + pub fn skip(sleep_hint: Duration) -> Self { + TickDecision { + execute_tasks: false, + sleep_hint: Some(sleep_hint), + } + } +} + +pub trait DistributedScheduler { + fn before_tick<'a>( + &'a mut self, + ) -> Pin> + 'a>>; + + fn after_tick<'a>( + &'a mut self, + scheduled_tasks: &'a mut BinaryHeap, + ) -> Pin> + 'a>>; + + fn shutdown<'a>(&'a mut self) -> Pin> + 'a>> { + Box::pin(async { Ok(()) }) + } +} + /// The default [`SchedulerBackend`](trait.SchedulerBackend.html). pub struct LocalSchedulerBackend {} diff --git a/src/beat/backend/redis.rs b/src/beat/backend/redis.rs new file mode 100644 index 00000000..516151e7 --- /dev/null +++ b/src/beat/backend/redis.rs @@ -0,0 +1,581 @@ +use super::{DistributedScheduler, TickDecision}; +use crate::beat::schedule::ScheduleDescriptor; +use crate::beat::scheduled_task::ScheduledTask; +use crate::error::BeatError; +use hostname::get as hostname_get; +use log::{info, warn}; +use redis::{AsyncCommands, Client, Script}; +use std::collections::{BinaryHeap, HashMap}; +use std::future::Future; +use std::pin::Pin; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use uuid::Uuid; + +const DEFAULT_KEY_PREFIX: &str = "celery_beat"; +const LOCK_RENEW_SCRIPT: &str = "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('PEXPIRE', KEYS[1], ARGV[2]) else return 0 end"; +const LOCK_RELEASE_SCRIPT: &str = "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end"; + +fn ensure_min_duration(duration: Duration) -> Duration { + if duration.is_zero() { + Duration::from_millis(1) + } else { + duration + } +} + +fn leader_sleep_hint(lock_renewal_interval: Duration) -> Duration { + let half = lock_renewal_interval.as_secs_f64() / 2.0; + if half < 0.001 { + Duration::from_millis(1) + } else { + Duration::from_secs_f64(half) + } +} + +fn generate_instance_id(prefix: &str) -> String { + let host = hostname_get() + .map(|s| s.to_string_lossy().into_owned()) + .unwrap_or_else(|_| "unknown-host".to_string()); + format!("{}:{}:{}", prefix, host, Uuid::new_v4()) +} + +fn system_time_to_epoch(time: SystemTime) -> u64 { + time.duration_since(UNIX_EPOCH) + .unwrap_or_else(|_| Duration::from_secs(0)) + .as_secs() +} + +fn epoch_to_system_time(epoch: u64) -> SystemTime { + UNIX_EPOCH + Duration::from_secs(epoch) +} + +#[derive(Clone)] +pub struct RedisBackendConfig { + redis_url: String, + key_prefix: String, + lock_timeout: Duration, + lock_renewal_interval: Duration, + follower_check_interval: Duration, + sync_interval: Duration, + follower_idle_sleep: Duration, + instance_id: Option, +} + +impl RedisBackendConfig { + pub fn new(redis_url: impl Into) -> Self { + Self { + redis_url: redis_url.into(), + key_prefix: DEFAULT_KEY_PREFIX.to_string(), + lock_timeout: Duration::from_secs(30), + lock_renewal_interval: Duration::from_secs(10), + follower_check_interval: Duration::from_secs(5), + sync_interval: Duration::from_secs(5), + follower_idle_sleep: Duration::from_millis(750), + instance_id: None, + } + } + + pub fn key_prefix(mut self, prefix: impl Into) -> Self { + self.key_prefix = prefix.into(); + self + } + + pub fn lock_timeout(mut self, timeout: Duration) -> Self { + self.lock_timeout = timeout; + self + } + + pub fn lock_renewal_interval(mut self, interval: Duration) -> Self { + self.lock_renewal_interval = interval; + self + } + + pub fn follower_check_interval(mut self, interval: Duration) -> Self { + self.follower_check_interval = interval; + self + } + + pub fn sync_interval(mut self, interval: Duration) -> Self { + self.sync_interval = interval; + self + } + + pub fn follower_idle_sleep(mut self, interval: Duration) -> Self { + self.follower_idle_sleep = interval; + self + } + + pub fn instance_id(mut self, id: impl Into) -> Self { + self.instance_id = Some(id.into()); + self + } + + pub fn resolve(self) -> ResolvedRedisBackendConfig { + let RedisBackendConfig { + redis_url, + key_prefix, + lock_timeout, + lock_renewal_interval, + follower_check_interval, + sync_interval, + follower_idle_sleep, + instance_id, + } = self; + + let instance_id = instance_id.unwrap_or_else(|| generate_instance_id(&key_prefix)); + + ResolvedRedisBackendConfig { + redis_url, + key_prefix: key_prefix.clone(), + lock_key: format!("{}:lock", key_prefix), + schedule_key: format!("{}:schedule", key_prefix), + instance_id, + lock_timeout, + lock_renewal_interval, + follower_check_interval, + sync_interval, + follower_idle_sleep, + } + } +} + +#[derive(Clone)] +pub struct ResolvedRedisBackendConfig { + pub redis_url: String, + pub key_prefix: String, + pub lock_key: String, + pub schedule_key: String, + pub instance_id: String, + pub lock_timeout: Duration, + pub lock_renewal_interval: Duration, + pub follower_check_interval: Duration, + pub sync_interval: Duration, + pub follower_idle_sleep: Duration, +} + +impl ResolvedRedisBackendConfig { + fn task_key(&self, name: &str) -> String { + format!("{}:task:{}", self.key_prefix, name) + } + + fn lock_ttl_millis(&self) -> usize { + self.lock_timeout.as_millis() as usize + } +} + +pub struct RedisSchedulerBackend { + config: ResolvedRedisBackendConfig, + client: Client, + state: BackendState, +} + +struct BackendState { + is_leader: bool, + last_lock_refresh: Option, + last_leader_attempt: Option, + last_sync: Option, + local_snapshot: HashMap, + pending_full_refresh: bool, +} + +#[derive(Clone, Debug, PartialEq)] +struct TaskState { + descriptor: ScheduleDescriptor, + next_run_at: SystemTime, + last_run_at: Option, + total_run_count: u32, +} + +impl RedisSchedulerBackend { + pub fn new(config: RedisBackendConfig) -> Result { + let resolved = config.resolve(); + let client = Client::open(resolved.redis_url.as_str()) + .map_err(|err| BeatError::RedisError(err.to_string()))?; + + Ok(Self { + config: resolved, + client, + state: BackendState { + is_leader: false, + last_lock_refresh: None, + last_leader_attempt: None, + last_sync: None, + local_snapshot: HashMap::new(), + pending_full_refresh: false, + }, + }) + } + + async fn get_connection(&self) -> Result { + self.client + .get_multiplexed_async_connection() + .await + .map_err(|err| BeatError::RedisError(err.to_string())) + } + + async fn try_acquire_lock(&mut self) -> Result { + let mut conn = self.get_connection().await?; + let result: Option = redis::cmd("SET") + .arg(&self.config.lock_key) + .arg(&self.config.instance_id) + .arg("NX") + .arg("PX") + .arg(self.config.lock_ttl_millis()) + .query_async(&mut conn) + .await + .map_err(|err| BeatError::RedisError(err.to_string()))?; + + if result.is_some() { + info!("Redis scheduler backend acquired leadership"); + self.state.last_lock_refresh = Some(Instant::now()); + self.state.is_leader = true; + self.state.pending_full_refresh = true; + Ok(true) + } else { + Ok(false) + } + } + + async fn renew_lock(&mut self) -> Result<(), BeatError> { + let mut conn = self.get_connection().await?; + let script = Script::new(LOCK_RENEW_SCRIPT); + let result: i32 = script + .key(&self.config.lock_key) + .arg(&self.config.instance_id) + .arg(self.config.lock_ttl_millis()) + .invoke_async(&mut conn) + .await + .map_err(|err| BeatError::RedisError(err.to_string()))?; + + if result == 1 { + self.state.last_lock_refresh = Some(Instant::now()); + Ok(()) + } else { + Err(BeatError::RedisError("lost leadership".into())) + } + } + + async fn release_lock(&mut self) -> Result<(), BeatError> { + let mut conn = self.get_connection().await?; + let script = Script::new(LOCK_RELEASE_SCRIPT); + let _: i32 = script + .key(&self.config.lock_key) + .arg(&self.config.instance_id) + .invoke_async(&mut conn) + .await + .map_err(|err| BeatError::RedisError(err.to_string()))?; + Ok(()) + } + + fn collect_task_state( + &self, + scheduled_tasks: &BinaryHeap, + ) -> (HashMap, Vec) { + let mut map = HashMap::new(); + let mut unsupported = Vec::new(); + + for task in scheduled_tasks.iter() { + let descriptor = match task.schedule.describe() { + Some(desc) => desc, + None => { + unsupported.push(task.name.clone()); + continue; + } + }; + + map.insert( + task.name.clone(), + TaskState { + descriptor, + next_run_at: task.next_call_at, + last_run_at: task.last_run_at, + total_run_count: task.total_run_count, + }, + ); + } + + (map, unsupported) + } + + async fn apply_remote_state( + &mut self, + scheduled_tasks: &mut BinaryHeap, + ) -> Result<(), BeatError> { + if scheduled_tasks.is_empty() { + self.state.local_snapshot.clear(); + return Ok(()); + } + + let mut tasks = Vec::with_capacity(scheduled_tasks.len()); + while let Some(task) = scheduled_tasks.pop() { + tasks.push(task); + } + + let mut conn = self.get_connection().await?; + for task in tasks.iter_mut() { + let key = self.config.task_key(&task.name); + let data: HashMap = conn + .hgetall(&key) + .await + .map_err(|err| BeatError::RedisError(err.to_string()))?; + + if data.is_empty() { + continue; + } + + if let Some(value) = data.get("last_run_at") { + if let Ok(epoch) = value.parse::() { + task.last_run_at = Some(epoch_to_system_time(epoch)); + } + } + if let Some(value) = data.get("next_run_at") { + if let Ok(epoch) = value.parse::() { + task.next_call_at = epoch_to_system_time(epoch); + } + } + if let Some(value) = data.get("total_run_count") { + if let Ok(count) = value.parse::() { + task.total_run_count = count; + } + } + } + + for task in tasks.into_iter() { + scheduled_tasks.push(task); + } + + Ok(()) + } + + async fn write_updates( + &mut self, + upserts: &HashMap, + deletes: &[String], + ) -> Result<(), BeatError> { + if upserts.is_empty() && deletes.is_empty() { + return Ok(()); + } + + let mut conn = self.get_connection().await?; + let mut pipe = redis::pipe(); + + for (name, state) in upserts { + let key = self.config.task_key(name); + let descriptor = serde_json::to_string(&state.descriptor) + .map_err(|err| BeatError::RedisError(err.to_string()))?; + + pipe.cmd("HSET") + .arg(&key) + .arg("descriptor") + .arg(descriptor) + .arg("task") + .arg(name) + .arg("total_run_count") + .arg(state.total_run_count) + .arg("next_run_at") + .arg(system_time_to_epoch(state.next_run_at)); + + if let Some(last_run) = state.last_run_at { + pipe.cmd("HSET") + .arg(&key) + .arg("last_run_at") + .arg(system_time_to_epoch(last_run)); + } + + pipe.cmd("ZADD") + .arg(&self.config.schedule_key) + .arg(system_time_to_epoch(state.next_run_at)) + .arg(&key); + } + + for name in deletes { + let key = self.config.task_key(name); + pipe.cmd("DEL").arg(&key); + pipe.cmd("ZREM").arg(&self.config.schedule_key).arg(&key); + } + + pipe.query_async::<()>(&mut conn) + .await + .map_err(|err| BeatError::RedisError(err.to_string()))?; + + Ok(()) + } +} + +impl super::SchedulerBackend for RedisSchedulerBackend { + fn should_sync(&self) -> bool { + false + } + + fn sync(&mut self, _scheduled_tasks: &mut BinaryHeap) -> Result<(), BeatError> { + Ok(()) + } + + fn as_distributed(&mut self) -> Option<&mut dyn DistributedScheduler> { + Some(self) + } +} + +impl DistributedScheduler for RedisSchedulerBackend { + fn before_tick<'a>( + &'a mut self, + ) -> Pin> + 'a>> { + Box::pin(async move { + let now = Instant::now(); + let leader_hint = leader_sleep_hint(self.config.lock_renewal_interval); + let follower_hint = ensure_min_duration(std::cmp::min( + self.config.follower_idle_sleep, + self.config.follower_check_interval, + )); + + if self.state.is_leader { + if self + .state + .last_lock_refresh + .map(|instant| now.duration_since(instant) >= self.config.lock_renewal_interval) + .unwrap_or(true) + { + if let Err(err) = self.renew_lock().await { + warn!("Redis scheduler backend failed to renew lock: {}", err); + self.state.is_leader = false; + return Ok(TickDecision::skip(follower_hint)); + } + } + Ok(TickDecision::execute_with_hint(leader_hint)) + } else { + if self + .state + .last_leader_attempt + .map(|instant| { + now.duration_since(instant) >= self.config.follower_check_interval + }) + .unwrap_or(true) + { + self.state.last_leader_attempt = Some(now); + if self.try_acquire_lock().await? { + return Ok(TickDecision::execute_with_hint(leader_hint)); + } + } + Ok(TickDecision::skip(follower_hint)) + } + }) + } + + fn after_tick<'a>( + &'a mut self, + scheduled_tasks: &'a mut BinaryHeap, + ) -> Pin> + 'a>> { + Box::pin(async move { + if !self.state.is_leader { + return Ok(()); + } + + if self.state.pending_full_refresh { + self.apply_remote_state(scheduled_tasks).await?; + self.state.pending_full_refresh = false; + } + + if self + .state + .last_sync + .map(|instant| instant.elapsed() < self.config.sync_interval) + .unwrap_or(false) + { + return Ok(()); + } + + let (current_state, unsupported) = self.collect_task_state(scheduled_tasks); + for name in unsupported { + warn!( + "Redis scheduler backend skipping task '{}' (unsupported schedule)", + name + ); + } + + let mut upserts = HashMap::new(); + for (name, state) in current_state.iter() { + match self.state.local_snapshot.get(name) { + Some(existing) if existing == state => {} + _ => { + upserts.insert(name.clone(), state.clone()); + } + } + } + + let mut deletes = Vec::new(); + for name in self.state.local_snapshot.keys() { + if !current_state.contains_key(name) { + deletes.push(name.clone()); + } + } + + self.write_updates(&upserts, &deletes).await?; + self.state.local_snapshot = current_state; + self.state.last_sync = Some(Instant::now()); + Ok(()) + }) + } + + fn shutdown<'a>(&'a mut self) -> Pin> + 'a>> { + Box::pin(async move { + if self.state.is_leader { + if let Err(err) = self.release_lock().await { + warn!("Redis scheduler backend failed to release lock: {}", err); + } + self.state.is_leader = false; + } + Ok(()) + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use uuid::Uuid; + + #[test] + fn resolve_applies_defaults() { + let config = RedisBackendConfig::new("redis://localhost:6379"); + let resolved = config.resolve(); + + assert_eq!(resolved.key_prefix, DEFAULT_KEY_PREFIX); + assert_eq!(resolved.lock_key, format!("{}:lock", DEFAULT_KEY_PREFIX)); + assert_eq!( + resolved.schedule_key, + format!("{}:schedule", DEFAULT_KEY_PREFIX) + ); + assert_eq!(resolved.lock_timeout, Duration::from_secs(30)); + assert_eq!(resolved.lock_renewal_interval, Duration::from_secs(10)); + assert_eq!(resolved.follower_check_interval, Duration::from_secs(5)); + assert!(resolved.instance_id.starts_with(DEFAULT_KEY_PREFIX)); + } + + #[tokio::test] + async fn lock_lifecycle_smoke() { + let url = + std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379/0".to_string()); + let prefix = format!("test_lock_{}", Uuid::new_v4()); + let config = RedisBackendConfig::new(&url).key_prefix(&prefix); + let mut backend = match RedisSchedulerBackend::new(config) { + Ok(backend) => backend, + Err(err) => { + eprintln!("Skipping Redis lock test: {err}"); + return; + } + }; + + match backend.try_acquire_lock().await { + Ok(true) => { + backend.renew_lock().await.expect("renew"); + backend.release_lock().await.expect("release"); + } + Ok(false) => { + eprintln!("Skipping Redis lock test: lock already held"); + } + Err(err) => { + eprintln!("Skipping Redis lock test: {err}"); + } + } + } +} diff --git a/src/beat/mod.rs b/src/beat/mod.rs index c54360ba..819e97fa 100644 --- a/src/beat/mod.rs +++ b/src/beat/mod.rs @@ -38,10 +38,12 @@ mod scheduler; pub use scheduler::Scheduler; mod backend; -pub use backend::{LocalSchedulerBackend, SchedulerBackend}; +pub use backend::{ + LocalSchedulerBackend, RedisBackendConfig, RedisSchedulerBackend, SchedulerBackend, +}; mod schedule; -pub use schedule::{CronSchedule, DeltaSchedule, Schedule}; +pub use schedule::{CronSchedule, DeltaSchedule, Schedule, ScheduleDescriptor}; mod scheduled_task; pub use scheduled_task::ScheduledTask; @@ -60,6 +62,15 @@ struct Config { } /// Used to create a [`Beat`] app with a custom configuration. +/// +/// # Choosing a backend +/// +/// - [`LocalSchedulerBackend`] keeps all scheduling state in memory and is intended for +/// single-instance deployments. +/// - [`RedisSchedulerBackend`] persists state in Redis and coordinates leadership across +/// multiple beat instances using Redis locks. When using the Redis backend you can further +/// tweak [`RedisBackendConfig`] to customise the lock key prefix, lock timeout, renewal +/// interval, follower polling interval, etc. pub struct BeatBuilder where Sb: SchedulerBackend, @@ -348,22 +359,61 @@ where async fn beat_loop(&mut self) -> Result<(), BeatError> { loop { - let next_tick_at = self.scheduler.tick().await?; + let mut sleep_hint = None; + let execute_tasks = { + if let Some(distributed) = self.scheduler_backend.as_distributed() { + let decision = distributed.before_tick().await?; + sleep_hint = decision.sleep_hint; + decision.execute_tasks + } else { + true + } + }; + + let next_tick_at = if execute_tasks { + self.scheduler.tick().await? + } else { + let fallback = sleep_hint.unwrap_or_else(|| Duration::from_secs(1)); + SystemTime::now() + fallback + }; + + let used_distributed = + if let Some(distributed) = self.scheduler_backend.as_distributed() { + distributed + .after_tick(self.scheduler.get_scheduled_tasks()) + .await?; + true + } else { + false + }; - if self.scheduler_backend.should_sync() { + if !used_distributed && self.scheduler_backend.should_sync() { self.scheduler_backend .sync(self.scheduler.get_scheduled_tasks())?; } let now = SystemTime::now(); - if now < next_tick_at { - let sleep_interval = next_tick_at.duration_since(now).expect( + let mut sleep_interval = if now < next_tick_at { + next_tick_at.duration_since(now).expect( "Unexpected error when unwrapping a SystemTime comparison that is not supposed to fail", - ); - let sleep_interval = match &self.max_sleep_duration { - Some(max_sleep_duration) => std::cmp::min(sleep_interval, *max_sleep_duration), - None => sleep_interval, - }; + ) + } else { + Duration::from_millis(0) + }; + + if !execute_tasks { + if let Some(hint) = sleep_hint { + sleep_interval = hint; + } + } else if let Some(hint) = sleep_hint { + sleep_interval = std::cmp::min(sleep_interval, hint); + } + + if let Some(max_sleep_duration) = &self.max_sleep_duration { + sleep_interval = std::cmp::min(sleep_interval, *max_sleep_duration); + } + + if sleep_interval > Duration::from_millis(0) { debug!("Now sleeping for {:?}", sleep_interval); time::sleep(sleep_interval).await; } diff --git a/src/beat/schedule/cron/mod.rs b/src/beat/schedule/cron/mod.rs index 4cfd7751..5b2686ac 100644 --- a/src/beat/schedule/cron/mod.rs +++ b/src/beat/schedule/cron/mod.rs @@ -3,9 +3,10 @@ //! [cron crate](https://crates.io/crates/cron). use chrono::{offset::Utc, TimeZone}; +use std::any::TypeId; use std::time::SystemTime; -use super::Schedule; +use super::{descriptor::CronDescriptor, Schedule}; use crate::error::ScheduleError; mod parsing; @@ -64,12 +65,20 @@ where impl Schedule for CronSchedule where - Z: TimeZone, + Z: TimeZone + 'static, { fn next_call_at(&self, _last_run_at: Option) -> Option { let now = self.time_zone.from_utc_datetime(&Utc::now().naive_utc()); self.next(now).map(SystemTime::from) } + + fn describe(&self) -> Option { + if TypeId::of::() == TypeId::of::() { + Some(super::ScheduleDescriptor::cron(self.snapshot())) + } else { + None + } + } } impl CronSchedule { @@ -116,6 +125,25 @@ impl CronSchedule { pub fn from_string(schedule: &str) -> Result { Self::from_string_with_time_zone(schedule, Utc) } + + pub fn describe(&self) -> CronDescriptor { + self.snapshot() + } +} + +impl CronSchedule +where + Z: TimeZone + 'static, +{ + fn snapshot(&self) -> CronDescriptor { + CronDescriptor { + minutes: self.minutes.to_vec(), + hours: self.hours.to_vec(), + month_days: self.month_days.to_vec(), + months: self.months.to_vec(), + week_days: self.week_days.to_vec(), + } + } } impl CronSchedule diff --git a/src/beat/schedule/cron/time_units.rs b/src/beat/schedule/cron/time_units.rs index a8d64ca2..c279a62a 100644 --- a/src/beat/schedule/cron/time_units.rs +++ b/src/beat/schedule/cron/time_units.rs @@ -36,6 +36,14 @@ impl Minutes { List(vec) => TimeUnitFieldIterator::from_vec(vec, start, Minutes::inclusive_max()), } } + + pub fn to_vec(&self) -> Vec { + use Minutes::*; + match self { + All => (Minutes::inclusive_min()..=Minutes::inclusive_max()).collect(), + List(vec) => vec.clone(), + } + } } impl TimeUnitField for Minutes { @@ -74,6 +82,14 @@ impl Hours { List(vec) => TimeUnitFieldIterator::from_vec(vec, start, Hours::inclusive_max()), } } + + pub fn to_vec(&self) -> Vec { + use Hours::*; + match self { + All => (Hours::inclusive_min()..=Hours::inclusive_max()).collect(), + List(vec) => vec.clone(), + } + } } impl TimeUnitField for Hours { @@ -112,6 +128,14 @@ impl WeekDays { List(vec) => vec.binary_search(&target).is_ok(), } } + + pub fn to_vec(&self) -> Vec { + use WeekDays::*; + match self { + All => (WeekDays::inclusive_min()..=WeekDays::inclusive_max()).collect(), + List(vec) => vec.clone(), + } + } } impl TimeUnitField for WeekDays { @@ -160,6 +184,14 @@ impl MonthDays { List(vec) => TimeUnitFieldIterator::from_vec(vec, start, stop), } } + + pub fn to_vec(&self) -> Vec { + use MonthDays::*; + match self { + All => (MonthDays::inclusive_min()..=MonthDays::inclusive_max()).collect(), + List(vec) => vec.clone(), + } + } } impl TimeUnitField for MonthDays { @@ -198,6 +230,14 @@ impl Months { List(vec) => TimeUnitFieldIterator::from_vec(vec, start, Months::inclusive_max()), } } + + pub fn to_vec(&self) -> Vec { + use Months::*; + match self { + All => (Months::inclusive_min()..=Months::inclusive_max()).collect(), + List(vec) => vec.clone(), + } + } } impl TimeUnitField for Months { diff --git a/src/beat/schedule/descriptor.rs b/src/beat/schedule/descriptor.rs new file mode 100644 index 00000000..17cc60c6 --- /dev/null +++ b/src/beat/schedule/descriptor.rs @@ -0,0 +1,106 @@ +use super::{CronSchedule, DeltaSchedule, Schedule}; +use crate::error::ScheduleError; +use serde::{Deserialize, Serialize}; +use std::time::Duration; + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub enum ScheduleDescriptor { + Cron(CronDescriptor), + Delta(DeltaDescriptor), +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct CronDescriptor { + pub minutes: Vec, + pub hours: Vec, + pub month_days: Vec, + pub months: Vec, + pub week_days: Vec, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct DeltaDescriptor { + pub interval_millis: u64, +} + +impl ScheduleDescriptor { + pub fn from_schedule(schedule: &dyn Schedule) -> Option { + schedule.describe() + } + + pub fn delta(interval: Duration) -> Self { + ScheduleDescriptor::Delta(DeltaDescriptor { + interval_millis: interval.as_millis() as u64, + }) + } + + pub fn cron(descriptor: CronDescriptor) -> Self { + ScheduleDescriptor::Cron(descriptor) + } + + pub fn to_schedule(&self) -> Result, ScheduleError> { + match self { + ScheduleDescriptor::Delta(desc) => Ok(Box::new(DeltaSchedule::new( + Duration::from_millis(desc.interval_millis), + ))), + ScheduleDescriptor::Cron(desc) => Ok(Box::new(desc.to_schedule()?)), + } + } +} + +impl CronDescriptor { + pub fn to_schedule(&self) -> Result, ScheduleError> { + CronSchedule::new( + self.minutes.clone(), + self.hours.clone(), + self.month_days.clone(), + self.months.clone(), + self.week_days.clone(), + ) + } +} + +impl DeltaDescriptor { + pub fn interval(&self) -> Duration { + Duration::from_millis(self.interval_millis) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn delta_descriptor_roundtrip() { + let schedule = DeltaSchedule::new(Duration::from_millis(250)); + let descriptor = ScheduleDescriptor::from_schedule(&schedule).expect("descriptor"); + + let restored = descriptor.to_schedule().expect("restored"); + let next_original = schedule.next_call_at(None).unwrap(); + let next_restored = restored.next_call_at(None).unwrap(); + + assert!( + next_restored + .duration_since(next_original) + .unwrap_or_else(|_| Duration::from_secs(0)) + < Duration::from_millis(2) + ); + } + + #[test] + fn cron_descriptor_roundtrip() { + let cron = CronSchedule::from_string("*/5 * * * *").expect("cron"); + let descriptor = ScheduleDescriptor::from_schedule(&cron).expect("descriptor"); + + let restored = descriptor.to_schedule().expect("restored"); + let next_original = cron.next_call_at(None).unwrap(); + let next_restored = restored.next_call_at(None).unwrap(); + + assert!( + next_restored + .duration_since(next_original) + .unwrap_or_else(|_| Duration::from_secs(0)) + < Duration::from_secs(60) + ); + } +} diff --git a/src/beat/schedule/mod.rs b/src/beat/schedule/mod.rs index 9d03c2a9..a0d6b167 100644 --- a/src/beat/schedule/mod.rs +++ b/src/beat/schedule/mod.rs @@ -3,18 +3,27 @@ //! These structs have not changed a lot compared to Python: in Python there are three //! different types of schedules: `schedule` (corresponding to [`DeltaSchedule`]), //! `crontab` (corresponding to [`CronSchedule`]), `solar` (not implemented yet). +use std::any::Any; use std::time::{Duration, SystemTime}; mod cron; pub use cron::CronSchedule; +mod descriptor; +pub use descriptor::ScheduleDescriptor; + /// The trait that all schedules implement. -pub trait Schedule { +pub trait Schedule: Any { /// Compute when a task should be executed again. /// If this method returns `None` then the task should /// never run again and it is safe to remove it from the /// list of scheduled tasks. fn next_call_at(&self, last_run_at: Option) -> Option; + + /// Describe this schedule so it can be serialized for persistence. + fn describe(&self) -> Option { + None + } } /// A schedule that can be used to execute tasks at regular intervals. @@ -29,6 +38,10 @@ impl DeltaSchedule { pub fn new(interval: Duration) -> DeltaSchedule { DeltaSchedule { interval } } + + pub fn interval(&self) -> Duration { + self.interval + } } impl Schedule for DeltaSchedule { @@ -42,4 +55,8 @@ impl Schedule for DeltaSchedule { None => Some(SystemTime::now()), } } + + fn describe(&self) -> Option { + Some(ScheduleDescriptor::delta(self.interval)) + } } diff --git a/src/error.rs b/src/error.rs index 13f89c21..e2b9819f 100644 --- a/src/error.rs +++ b/src/error.rs @@ -50,6 +50,10 @@ pub enum BeatError { /// An error with a task schedule. #[error("task schedule error")] ScheduleError(#[from] ScheduleError), + + /// Redis-related error. + #[error("redis error: {0}")] + RedisError(String), } /// Errors that are related to task schedules. diff --git a/uv.lock b/uv.lock new file mode 100644 index 00000000..b40695dc --- /dev/null +++ b/uv.lock @@ -0,0 +1,221 @@ +version = 1 +revision = 2 +requires-python = ">=3.10" + +[[package]] +name = "amqp" +version = "5.3.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "vine" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/79/fc/ec94a357dfc6683d8c86f8b4cfa5416a4c36b28052ec8260c77aca96a443/amqp-5.3.1.tar.gz", hash = "sha256:cddc00c725449522023bad949f70fff7b48f0b1ade74d170a6f10ab044739432", size = 129013, upload-time = "2024-11-12T19:55:44.051Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/26/99/fc813cd978842c26c82534010ea849eee9ab3a13ea2b74e95cb9c99e747b/amqp-5.3.1-py3-none-any.whl", hash = "sha256:43b3319e1b4e7d1251833a93d672b4af1e40f3d632d479b98661a95f117880a2", size = 50944, upload-time = "2024-11-12T19:55:41.782Z" }, +] + +[[package]] +name = "async-timeout" +version = "5.0.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a5/ae/136395dfbfe00dfc94da3f3e136d0b13f394cba8f4841120e34226265780/async_timeout-5.0.1.tar.gz", hash = "sha256:d9321a7a3d5a6a5e187e824d2fa0793ce379a202935782d555d6e9d2735677d3", size = 9274, upload-time = "2024-11-06T16:41:39.6Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fe/ba/e2081de779ca30d473f21f5b30e0e737c438205440784c7dfc81efc2b029/async_timeout-5.0.1-py3-none-any.whl", hash = "sha256:39e3809566ff85354557ec2398b55e096c8364bacac9405a7a1fa429e77fe76c", size = 6233, upload-time = "2024-11-06T16:41:37.9Z" }, +] + +[[package]] +name = "billiard" +version = "4.2.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/b9/6a/1405343016bce8354b29d90aad6b0bf6485b5e60404516e4b9a3a9646cf0/billiard-4.2.2.tar.gz", hash = "sha256:e815017a062b714958463e07ba15981d802dc53d41c5b69d28c5a7c238f8ecf3", size = 155592, upload-time = "2025-09-20T14:44:40.456Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a6/80/ef8dff49aae0e4430f81842f7403e14e0ca59db7bbaf7af41245b67c6b25/billiard-4.2.2-py3-none-any.whl", hash = "sha256:4bc05dcf0d1cc6addef470723aac2a6232f3c7ed7475b0b580473a9145829457", size = 86896, upload-time = "2025-09-20T14:44:39.157Z" }, +] + +[[package]] +name = "celery" +version = "5.5.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "billiard" }, + { name = "click" }, + { name = "click-didyoumean" }, + { name = "click-plugins" }, + { name = "click-repl" }, + { name = "kombu" }, + { name = "python-dateutil" }, + { name = "vine" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/bb/7d/6c289f407d219ba36d8b384b42489ebdd0c84ce9c413875a8aae0c85f35b/celery-5.5.3.tar.gz", hash = "sha256:6c972ae7968c2b5281227f01c3a3f984037d21c5129d07bf3550cc2afc6b10a5", size = 1667144, upload-time = "2025-06-01T11:08:12.563Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c9/af/0dcccc7fdcdf170f9a1585e5e96b6fb0ba1749ef6be8c89a6202284759bd/celery-5.5.3-py3-none-any.whl", hash = "sha256:0b5761a07057acee94694464ca482416b959568904c9dfa41ce8413a7d65d525", size = 438775, upload-time = "2025-06-01T11:08:09.94Z" }, +] + +[[package]] +name = "celery-rs-python-examples" +version = "0.1.0" +source = { virtual = "." } +dependencies = [ + { name = "celery" }, + { name = "redis" }, +] + +[package.metadata] +requires-dist = [ + { name = "celery", specifier = ">=5.3" }, + { name = "redis", specifier = ">=4.5" }, +] + +[[package]] +name = "click" +version = "8.3.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/46/61/de6cd827efad202d7057d93e0fed9294b96952e188f7384832791c7b2254/click-8.3.0.tar.gz", hash = "sha256:e7b8232224eba16f4ebe410c25ced9f7875cb5f3263ffc93cc3e8da705e229c4", size = 276943, upload-time = "2025-09-18T17:32:23.696Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/db/d3/9dcc0f5797f070ec8edf30fbadfb200e71d9db6b84d211e3b2085a7589a0/click-8.3.0-py3-none-any.whl", hash = "sha256:9b9f285302c6e3064f4330c05f05b81945b2a39544279343e6e7c5f27a9baddc", size = 107295, upload-time = "2025-09-18T17:32:22.42Z" }, +] + +[[package]] +name = "click-didyoumean" +version = "0.3.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "click" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/30/ce/217289b77c590ea1e7c24242d9ddd6e249e52c795ff10fac2c50062c48cb/click_didyoumean-0.3.1.tar.gz", hash = "sha256:4f82fdff0dbe64ef8ab2279bd6aa3f6a99c3b28c05aa09cbfc07c9d7fbb5a463", size = 3089, upload-time = "2024-03-24T08:22:07.499Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/1b/5b/974430b5ffdb7a4f1941d13d83c64a0395114503cc357c6b9ae4ce5047ed/click_didyoumean-0.3.1-py3-none-any.whl", hash = "sha256:5c4bb6007cfea5f2fd6583a2fb6701a22a41eb98957e63d0fac41c10e7c3117c", size = 3631, upload-time = "2024-03-24T08:22:06.356Z" }, +] + +[[package]] +name = "click-plugins" +version = "1.1.1.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "click" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c3/a4/34847b59150da33690a36da3681d6bbc2ec14ee9a846bc30a6746e5984e4/click_plugins-1.1.1.2.tar.gz", hash = "sha256:d7af3984a99d243c131aa1a828331e7630f4a88a9741fd05c927b204bcf92261", size = 8343, upload-time = "2025-06-25T00:47:37.555Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3d/9a/2abecb28ae875e39c8cad711eb1186d8d14eab564705325e77e4e6ab9ae5/click_plugins-1.1.1.2-py2.py3-none-any.whl", hash = "sha256:008d65743833ffc1f5417bf0e78e8d2c23aab04d9745ba817bd3e71b0feb6aa6", size = 11051, upload-time = "2025-06-25T00:47:36.731Z" }, +] + +[[package]] +name = "click-repl" +version = "0.3.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "click" }, + { name = "prompt-toolkit" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/cb/a2/57f4ac79838cfae6912f997b4d1a64a858fb0c86d7fcaae6f7b58d267fca/click-repl-0.3.0.tar.gz", hash = "sha256:17849c23dba3d667247dc4defe1757fff98694e90fe37474f3feebb69ced26a9", size = 10449, upload-time = "2023-06-15T12:43:51.141Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/52/40/9d857001228658f0d59e97ebd4c346fe73e138c6de1bce61dc568a57c7f8/click_repl-0.3.0-py3-none-any.whl", hash = "sha256:fb7e06deb8da8de86180a33a9da97ac316751c094c6899382da7feeeeb51b812", size = 10289, upload-time = "2023-06-15T12:43:48.626Z" }, +] + +[[package]] +name = "colorama" +version = "0.4.6" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, +] + +[[package]] +name = "kombu" +version = "5.5.4" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "amqp" }, + { name = "packaging" }, + { name = "tzdata" }, + { name = "vine" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/0f/d3/5ff936d8319ac86b9c409f1501b07c426e6ad41966fedace9ef1b966e23f/kombu-5.5.4.tar.gz", hash = "sha256:886600168275ebeada93b888e831352fe578168342f0d1d5833d88ba0d847363", size = 461992, upload-time = "2025-06-01T10:19:22.281Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ef/70/a07dcf4f62598c8ad579df241af55ced65bed76e42e45d3c368a6d82dbc1/kombu-5.5.4-py3-none-any.whl", hash = "sha256:a12ed0557c238897d8e518f1d1fdf84bd1516c5e305af2dacd85c2015115feb8", size = 210034, upload-time = "2025-06-01T10:19:20.436Z" }, +] + +[[package]] +name = "packaging" +version = "25.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a1/d4/1fc4078c65507b51b96ca8f8c3ba19e6a61c8253c72794544580a7b6c24d/packaging-25.0.tar.gz", hash = "sha256:d443872c98d677bf60f6a1f2f8c1cb748e8fe762d2bf9d3148b5599295b0fc4f", size = 165727, upload-time = "2025-04-19T11:48:59.673Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/20/12/38679034af332785aac8774540895e234f4d07f7545804097de4b666afd8/packaging-25.0-py3-none-any.whl", hash = "sha256:29572ef2b1f17581046b3a2227d5c611fb25ec70ca1ba8554b24b0e69331a484", size = 66469, upload-time = "2025-04-19T11:48:57.875Z" }, +] + +[[package]] +name = "prompt-toolkit" +version = "3.0.52" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "wcwidth" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a1/96/06e01a7b38dce6fe1db213e061a4602dd6032a8a97ef6c1a862537732421/prompt_toolkit-3.0.52.tar.gz", hash = "sha256:28cde192929c8e7321de85de1ddbe736f1375148b02f2e17edd840042b1be855", size = 434198, upload-time = "2025-08-27T15:24:02.057Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/84/03/0d3ce49e2505ae70cf43bc5bb3033955d2fc9f932163e84dc0779cc47f48/prompt_toolkit-3.0.52-py3-none-any.whl", hash = "sha256:9aac639a3bbd33284347de5ad8d68ecc044b91a762dc39b7c21095fcd6a19955", size = 391431, upload-time = "2025-08-27T15:23:59.498Z" }, +] + +[[package]] +name = "python-dateutil" +version = "2.9.0.post0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "six" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/66/c0/0c8b6ad9f17a802ee498c46e004a0eb49bc148f2fd230864601a86dcf6db/python-dateutil-2.9.0.post0.tar.gz", hash = "sha256:37dd54208da7e1cd875388217d5e00ebd4179249f90fb72437e91a35459a0ad3", size = 342432, upload-time = "2024-03-01T18:36:20.211Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" }, +] + +[[package]] +name = "redis" +version = "6.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "async-timeout", marker = "python_full_version < '3.11.3'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/0d/d6/e8b92798a5bd67d659d51a18170e91c16ac3b59738d91894651ee255ed49/redis-6.4.0.tar.gz", hash = "sha256:b01bc7282b8444e28ec36b261df5375183bb47a07eb9c603f284e89cbc5ef010", size = 4647399, upload-time = "2025-08-07T08:10:11.441Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e8/02/89e2ed7e85db6c93dfa9e8f691c5087df4e3551ab39081a4d7c6d1f90e05/redis-6.4.0-py3-none-any.whl", hash = "sha256:f0544fa9604264e9464cdf4814e7d4830f74b165d52f2a330a760a88dd248b7f", size = 279847, upload-time = "2025-08-07T08:10:09.84Z" }, +] + +[[package]] +name = "six" +version = "1.17.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/94/e7/b2c673351809dca68a0e064b6af791aa332cf192da575fd474ed7d6f16a2/six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81", size = 34031, upload-time = "2024-12-04T17:35:28.174Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" }, +] + +[[package]] +name = "tzdata" +version = "2025.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/95/32/1a225d6164441be760d75c2c42e2780dc0873fe382da3e98a2e1e48361e5/tzdata-2025.2.tar.gz", hash = "sha256:b60a638fcc0daffadf82fe0f57e53d06bdec2f36c4df66280ae79bce6bd6f2b9", size = 196380, upload-time = "2025-03-23T13:54:43.652Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/5c/23/c7abc0ca0a1526a0774eca151daeb8de62ec457e77262b66b359c3c7679e/tzdata-2025.2-py2.py3-none-any.whl", hash = "sha256:1a403fada01ff9221ca8044d701868fa132215d84beb92242d9acd2147f667a8", size = 347839, upload-time = "2025-03-23T13:54:41.845Z" }, +] + +[[package]] +name = "vine" +version = "5.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/bd/e4/d07b5f29d283596b9727dd5275ccbceb63c44a1a82aa9e4bfd20426762ac/vine-5.1.0.tar.gz", hash = "sha256:8b62e981d35c41049211cf62a0a1242d8c1ee9bd15bb196ce38aefd6799e61e0", size = 48980, upload-time = "2023-11-05T08:46:53.857Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/03/ff/7c0c86c43b3cbb927e0ccc0255cb4057ceba4799cd44ae95174ce8e8b5b2/vine-5.1.0-py3-none-any.whl", hash = "sha256:40fdf3c48b2cfe1c38a49e9ae2da6fda88e4794c810050a728bd7413811fb1dc", size = 9636, upload-time = "2023-11-05T08:46:51.205Z" }, +] + +[[package]] +name = "wcwidth" +version = "0.2.14" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/24/30/6b0809f4510673dc723187aeaf24c7f5459922d01e2f794277a3dfb90345/wcwidth-0.2.14.tar.gz", hash = "sha256:4d478375d31bc5395a3c55c40ccdf3354688364cd61c4f6adacaa9215d0b3605", size = 102293, upload-time = "2025-09-22T16:29:53.023Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/af/b5/123f13c975e9f27ab9c0770f514345bd406d0e8d3b7a0723af9d43f710af/wcwidth-0.2.14-py2.py3-none-any.whl", hash = "sha256:a7bb560c8aee30f9957e5f9895805edd20602f2d7f720186dfd906e82b4982e1", size = 37286, upload-time = "2025-09-22T16:29:51.641Z" }, +]