Add Redis scheduler backend with distributed leadership support

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

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"

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.<br/>
@@ -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
 

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,

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<i32> {
+    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(())
+}

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
 

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<ScheduledTask>) -> 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<Duration>,
+}
+
+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<Box<dyn Future<Output = Result<TickDecision, BeatError>> + 'a>>;
+
+    fn after_tick<'a>(
+        &'a mut self,
+        scheduled_tasks: &'a mut BinaryHeap<ScheduledTask>,
+    ) -> Pin<Box<dyn Future<Output = Result<(), BeatError>> + 'a>>;
+
+    fn shutdown<'a>(&'a mut self) -> Pin<Box<dyn Future<Output = Result<(), BeatError>> + 'a>> {
+        Box::pin(async { Ok(()) })
+    }
+}
+
 /// The default [`SchedulerBackend`](trait.SchedulerBackend.html).
 pub struct LocalSchedulerBackend {}