From 623f6ca3ab01b5c1f77a18a5cce87aee8dfcd66b Mon Sep 17 00:00:00 2001 From: dragon Date: Wed, 1 Oct 2025 16:35:09 +0800 Subject: [PATCH 1/4] Add Redis-backed distributed scheduler backend Introduces a Redis-powered distributed scheduler backend (RedisSchedulerBackend) with leader election and state synchronization for scheduled tasks. Updates the beat loop to support distributed backends, adds schedule descriptors for serialization, and extends cron/interval schedule types for persistence. Includes a new example (redis_beat.rs) and updates documentation for usage. --- Cargo.toml | 5 +- README.md | 12 + examples/celery_app.rs | 4 +- examples/redis_beat.rs | 46 +++ src/beat/backend.rs | 50 ++- src/beat/backend/redis.rs | 558 +++++++++++++++++++++++++++ src/beat/mod.rs | 63 ++- src/beat/schedule/cron/mod.rs | 32 +- src/beat/schedule/cron/time_units.rs | 40 ++ src/beat/schedule/descriptor.rs | 106 +++++ src/beat/schedule/mod.rs | 19 +- src/error.rs | 4 + 12 files changed, 921 insertions(+), 18 deletions(-) create mode 100644 examples/redis_beat.rs create mode 100644 src/beat/backend/redis.rs create mode 100644 src/beat/schedule/descriptor.rs 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..326f1a9b 100644 --- a/README.md +++ b/README.md @@ -140,6 +140,18 @@ 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. + ## Road map and current state ✅ = Supported and mostly stable, although there may be a few incomplete features.
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/src/beat/backend.rs b/src/beat/backend.rs index 82975bb3..c0cd5348 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,56 @@ 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 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..ae83e9ee --- /dev/null +++ b/src/beat/backend/redis.rs @@ -0,0 +1,558 @@ +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 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(); + 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(self.config.follower_idle_sleep)); + } + } + Ok(TickDecision::execute()) + } 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()); + } + } + Ok(TickDecision::skip(self.config.follower_idle_sleep)) + } + }) + } + + 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..46b6e335 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; @@ -348,22 +350,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. From b47c9caea98a7a9e9445cfe74ef4742befd18e53 Mon Sep 17 00:00:00 2001 From: dragon Date: Wed, 1 Oct 2025 17:09:00 +0800 Subject: [PATCH 2/4] Improve Redis beat backend sleep hint logic Added leader and follower sleep hint logic to the RedisSchedulerBackend, ensuring more responsive failover and efficient scheduling. Updated documentation in README and beat module to clarify backend selection and failover testing. Also added a new TickDecision::execute_with_hint method for more flexible scheduling. --- README.md | 24 ++++++++++++++++-------- src/beat/backend.rs | 7 +++++++ src/beat/backend/redis.rs | 31 +++++++++++++++++++++++++++---- src/beat/mod.rs | 9 +++++++++ 4 files changed, 59 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 326f1a9b..19587a85 100644 --- a/README.md +++ b/README.md @@ -152,6 +152,14 @@ 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.
@@ -160,31 +168,31 @@ take over automatically if the leader disconnects. ### 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/src/beat/backend.rs b/src/beat/backend.rs index c0cd5348..ec0c9bb8 100644 --- a/src/beat/backend.rs +++ b/src/beat/backend.rs @@ -54,6 +54,13 @@ impl TickDecision { } } + 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, diff --git a/src/beat/backend/redis.rs b/src/beat/backend/redis.rs index ae83e9ee..516151e7 100644 --- a/src/beat/backend/redis.rs +++ b/src/beat/backend/redis.rs @@ -15,6 +15,23 @@ 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()) @@ -405,6 +422,12 @@ impl DistributedScheduler for RedisSchedulerBackend { ) -> 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 @@ -415,10 +438,10 @@ impl DistributedScheduler for RedisSchedulerBackend { 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(self.config.follower_idle_sleep)); + return Ok(TickDecision::skip(follower_hint)); } } - Ok(TickDecision::execute()) + Ok(TickDecision::execute_with_hint(leader_hint)) } else { if self .state @@ -430,10 +453,10 @@ impl DistributedScheduler for RedisSchedulerBackend { { self.state.last_leader_attempt = Some(now); if self.try_acquire_lock().await? { - return Ok(TickDecision::execute()); + return Ok(TickDecision::execute_with_hint(leader_hint)); } } - Ok(TickDecision::skip(self.config.follower_idle_sleep)) + Ok(TickDecision::skip(follower_hint)) } }) } diff --git a/src/beat/mod.rs b/src/beat/mod.rs index 46b6e335..819e97fa 100644 --- a/src/beat/mod.rs +++ b/src/beat/mod.rs @@ -62,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, From 74f06a7f04a75a6189383e87438907bed3cf7f4e Mon Sep 17 00:00:00 2001 From: dragon Date: Wed, 1 Oct 2025 17:23:07 +0800 Subject: [PATCH 3/4] Add project metadata and dependencies Added [project] section to pyproject.toml with project name, version, Python requirement, and dependencies on celery and redis. This sets up the initial project configuration for dependency management. --- pyproject.toml | 9 ++ uv.lock | 221 +++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 230 insertions(+) create mode 100644 uv.lock 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/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" }, +] From 8e12f16de3785f36d2aef3b3ccf13b7fd090e4be Mon Sep 17 00:00:00 2001 From: dragon Date: Wed, 1 Oct 2025 17:32:09 +0800 Subject: [PATCH 4/4] Add Redis-backed distributed scheduler backend Introduced a Redis-based scheduler backend with configurable lock management and task state persistence. Added documentation and examples for multi-instance beat deployments. --- CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) 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