From e3188f8802378e6875d60533f4731ce1827014f9 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 27 Jun 2026 13:06:10 +0530 Subject: [PATCH 1/5] feat(core): periodic list/delete/pause storage ops list_periodic, delete_periodic, and set_periodic_enabled across SQLite/Postgres/Redis. Pause/resume reuses the existing enabled flag (no schema change); get_due_periodic already filters it. Redis gains a periodic:all index for listing. --- crates/taskito-core/src/storage/mod.rs | 24 ++++++ .../src/storage/postgres/periodic.rs | 32 ++++++++ .../src/storage/redis_backend/periodic.rs | 75 +++++++++++++++++++ .../src/storage/sqlite/periodic.rs | 32 ++++++++ .../taskito-core/src/storage/sqlite/tests.rs | 65 ++++++++++++++++ crates/taskito-core/src/storage/traits.rs | 7 ++ .../taskito-core/tests/rust/storage_tests.rs | 52 +++++++++++++ 7 files changed, 287 insertions(+) diff --git a/crates/taskito-core/src/storage/mod.rs b/crates/taskito-core/src/storage/mod.rs index 2684a4a9..7c8a3f46 100644 --- a/crates/taskito-core/src/storage/mod.rs +++ b/crates/taskito-core/src/storage/mod.rs @@ -306,6 +306,21 @@ macro_rules! impl_storage { ) -> $crate::error::Result<()> { self.update_periodic_schedule(name, last_run, next_run) } + fn list_periodic( + &self, + ) -> $crate::error::Result> { + self.list_periodic() + } + fn delete_periodic(&self, name: &str) -> $crate::error::Result { + self.delete_periodic(name) + } + fn set_periodic_enabled( + &self, + name: &str, + enabled: bool, + ) -> $crate::error::Result { + self.set_periodic_enabled(name, enabled) + } fn record_metric( &self, task_name: &str, @@ -800,6 +815,15 @@ impl Storage for StorageBackend { fn update_periodic_schedule(&self, name: &str, last_run: i64, next_run: i64) -> Result<()> { delegate!(self, update_periodic_schedule, name, last_run, next_run) } + fn list_periodic(&self) -> Result> { + delegate!(self, list_periodic) + } + fn delete_periodic(&self, name: &str) -> Result { + delegate!(self, delete_periodic, name) + } + fn set_periodic_enabled(&self, name: &str, enabled: bool) -> Result { + delegate!(self, set_periodic_enabled, name, enabled) + } fn record_metric( &self, task_name: &str, diff --git a/crates/taskito-core/src/storage/postgres/periodic.rs b/crates/taskito-core/src/storage/postgres/periodic.rs index 6a825283..f8a4e919 100644 --- a/crates/taskito-core/src/storage/postgres/periodic.rs +++ b/crates/taskito-core/src/storage/postgres/periodic.rs @@ -46,4 +46,36 @@ impl PostgresStorage { Ok(()) } + + /// List all registered periodic tasks, enabled or paused. + pub fn list_periodic(&self) -> Result> { + let mut conn = self.conn()?; + + let rows = periodic_tasks::table + .select(PeriodicTaskRow::as_select()) + .load(&mut conn)?; + + Ok(rows) + } + + /// Remove a periodic task. Returns false if no task had that name. + pub fn delete_periodic(&self, name: &str) -> Result { + let mut conn = self.conn()?; + + let affected = diesel::delete(periodic_tasks::table.find(name)).execute(&mut conn)?; + + Ok(affected > 0) + } + + /// Pause (false) or resume (true) a periodic task by toggling `enabled`. + /// Returns false if no task had that name. + pub fn set_periodic_enabled(&self, name: &str, enabled: bool) -> Result { + let mut conn = self.conn()?; + + let affected = diesel::update(periodic_tasks::table.find(name)) + .set(periodic_tasks::enabled.eq(enabled)) + .execute(&mut conn)?; + + Ok(affected > 0) + } } diff --git a/crates/taskito-core/src/storage/redis_backend/periodic.rs b/crates/taskito-core/src/storage/redis_backend/periodic.rs index 013fa4ae..299d97f4 100644 --- a/crates/taskito-core/src/storage/redis_backend/periodic.rs +++ b/crates/taskito-core/src/storage/redis_backend/periodic.rs @@ -57,9 +57,11 @@ impl RedisStorage { let pkey = self.key(&["periodic", task.name]); let due_key = self.key(&["periodic", "due"]); + let all_key = self.key(&["periodic", "all"]); let pipe = &mut redis::pipe(); pipe.set(&pkey, &json); + pipe.sadd(&all_key, task.name); if entry.enabled { pipe.zadd(&due_key, task.name, task.next_run as f64); } @@ -115,4 +117,77 @@ impl RedisStorage { Ok(()) } + + /// List all registered periodic tasks, enabled or paused. + pub fn list_periodic(&self) -> Result> { + let mut conn = self.conn()?; + let all_key = self.key(&["periodic", "all"]); + + let names: Vec = conn.smembers(&all_key).map_err(map_err)?; + + let mut rows = Vec::new(); + for name in names { + let pkey = self.key(&["periodic", &name]); + let data: Option = conn.get(&pkey).map_err(map_err)?; + match data { + Some(d) => { + let entry: PeriodicEntry = + serde_json::from_str(&d).map_err(|e| QueueError::Other(e.to_string()))?; + rows.push(PeriodicTaskRow::from(entry)); + } + // Stale index entry: the task key is gone, so self-heal. + None => { + let _removed: i64 = conn.srem(&all_key, &name).map_err(map_err)?; + } + } + } + + Ok(rows) + } + + /// Remove a periodic task. Returns false if no task had that name. + pub fn delete_periodic(&self, name: &str) -> Result { + let mut conn = self.conn()?; + let pkey = self.key(&["periodic", name]); + let due_key = self.key(&["periodic", "due"]); + let all_key = self.key(&["periodic", "all"]); + + let pipe = &mut redis::pipe(); + pipe.del(&pkey); + pipe.zrem(&due_key, name); + pipe.srem(&all_key, name); + let (deleted, _zrem, _srem): (i64, i64, i64) = pipe.query(&mut conn).map_err(map_err)?; + + Ok(deleted > 0) + } + + /// Pause (false) or resume (true) a periodic task by toggling `enabled`. + /// Returns false if no task had that name. + pub fn set_periodic_enabled(&self, name: &str, enabled: bool) -> Result { + let mut conn = self.conn()?; + let pkey = self.key(&["periodic", name]); + + let data: Option = conn.get(&pkey).map_err(map_err)?; + let Some(d) = data else { + return Ok(false); + }; + + let mut entry: PeriodicEntry = + serde_json::from_str(&d).map_err(|e| QueueError::Other(e.to_string()))?; + entry.enabled = enabled; + + let json = serde_json::to_string(&entry).map_err(|e| QueueError::Other(e.to_string()))?; + let due_key = self.key(&["periodic", "due"]); + + let pipe = &mut redis::pipe(); + pipe.set(&pkey, &json); + if enabled { + pipe.zadd(&due_key, name, entry.next_run as f64); + } else { + pipe.zrem(&due_key, name); + } + pipe.query::<()>(&mut conn).map_err(map_err)?; + + Ok(true) + } } diff --git a/crates/taskito-core/src/storage/sqlite/periodic.rs b/crates/taskito-core/src/storage/sqlite/periodic.rs index d6c5ff00..189d9a10 100644 --- a/crates/taskito-core/src/storage/sqlite/periodic.rs +++ b/crates/taskito-core/src/storage/sqlite/periodic.rs @@ -43,4 +43,36 @@ impl SqliteStorage { Ok(()) } + + /// List all registered periodic tasks, enabled or paused. + pub fn list_periodic(&self) -> Result> { + let mut conn = self.conn()?; + + let rows = periodic_tasks::table + .select(PeriodicTaskRow::as_select()) + .load(&mut conn)?; + + Ok(rows) + } + + /// Remove a periodic task. Returns false if no task had that name. + pub fn delete_periodic(&self, name: &str) -> Result { + let mut conn = self.conn()?; + + let affected = diesel::delete(periodic_tasks::table.find(name)).execute(&mut conn)?; + + Ok(affected > 0) + } + + /// Pause (false) or resume (true) a periodic task by toggling `enabled`. + /// Returns false if no task had that name. + pub fn set_periodic_enabled(&self, name: &str, enabled: bool) -> Result { + let mut conn = self.conn()?; + + let affected = diesel::update(periodic_tasks::table.find(name)) + .set(periodic_tasks::enabled.eq(enabled)) + .execute(&mut conn)?; + + Ok(affected > 0) + } } diff --git a/crates/taskito-core/src/storage/sqlite/tests.rs b/crates/taskito-core/src/storage/sqlite/tests.rs index a6ed74bf..c69cc1b6 100644 --- a/crates/taskito-core/src/storage/sqlite/tests.rs +++ b/crates/taskito-core/src/storage/sqlite/tests.rs @@ -1241,3 +1241,68 @@ fn test_dlq_retry_count_round_trip() { let dead2 = storage.list_dead(10, 0).unwrap(); assert_eq!(dead2[0].dlq_retry_count, 1); } + +fn make_periodic(name: &str, next_run: i64) -> crate::storage::models::NewPeriodicTaskRow<'_> { + crate::storage::models::NewPeriodicTaskRow { + name, + task_name: "periodic_task", + cron_expr: "* * * * *", + args: None, + kwargs: None, + queue: "default", + enabled: true, + next_run, + timezone: None, + } +} + +#[test] +fn test_periodic_pause_resume_and_delete() { + let storage = test_storage(); + let now = now_millis(); + let past = now - 1000; + + storage + .register_periodic(&make_periodic("alpha", past)) + .unwrap(); + storage + .register_periodic(&make_periodic("beta", past)) + .unwrap(); + + // Both registered tasks are listed. + let listed = storage.list_periodic().unwrap(); + assert_eq!(listed.len(), 2); + assert!(listed.iter().all(|row| row.enabled)); + + // Both are due and fire. + assert_eq!(storage.get_due_periodic(now).unwrap().len(), 2); + + // Pausing "alpha" toggles enabled off and stops it firing. + assert!(storage.set_periodic_enabled("alpha", false).unwrap()); + let due_names: Vec = storage + .get_due_periodic(now) + .unwrap() + .into_iter() + .map(|row| row.name) + .collect(); + assert_eq!(due_names, vec!["beta".to_string()]); + + // But it is still listed (paused, not removed). + assert_eq!(storage.list_periodic().unwrap().len(), 2); + + // Resuming brings it back into the due set. + assert!(storage.set_periodic_enabled("alpha", true).unwrap()); + assert_eq!(storage.get_due_periodic(now).unwrap().len(), 2); + + // Deleting removes it from the listing. + assert!(storage.delete_periodic("alpha").unwrap()); + let remaining = storage.list_periodic().unwrap(); + assert_eq!(remaining.len(), 1); + assert_eq!(remaining[0].name, "beta"); + + // A second delete reports nothing removed. + assert!(!storage.delete_periodic("alpha").unwrap()); + + // Toggling an unknown task reports nothing changed. + assert!(!storage.set_periodic_enabled("ghost", false).unwrap()); +} diff --git a/crates/taskito-core/src/storage/traits.rs b/crates/taskito-core/src/storage/traits.rs index f8237a72..34b9696e 100644 --- a/crates/taskito-core/src/storage/traits.rs +++ b/crates/taskito-core/src/storage/traits.rs @@ -102,6 +102,13 @@ pub trait Storage: Send + Sync + Clone { fn register_periodic(&self, task: &NewPeriodicTaskRow) -> Result<()>; fn get_due_periodic(&self, now: i64) -> Result>; fn update_periodic_schedule(&self, name: &str, last_run: i64, next_run: i64) -> Result<()>; + /// All registered periodic tasks, enabled or paused. + fn list_periodic(&self) -> Result>; + /// Remove a periodic task. Returns false if no task had that name. + fn delete_periodic(&self, name: &str) -> Result; + /// Pause (false) or resume (true) a periodic task by toggling `enabled`. + /// Returns false if no task had that name. + fn set_periodic_enabled(&self, name: &str, enabled: bool) -> Result; // ── Metrics operations ────────────────────────────────────────── diff --git a/crates/taskito-core/tests/rust/storage_tests.rs b/crates/taskito-core/tests/rust/storage_tests.rs index baea7540..9cdee058 100644 --- a/crates/taskito-core/tests/rust/storage_tests.rs +++ b/crates/taskito-core/tests/rust/storage_tests.rs @@ -626,6 +626,57 @@ fn test_archived_job_payload_resolves(s: &impl Storage) { assert_eq!(row.result, Some(vec![0x11, 0x22])); } +fn due_periodic_names(s: &impl Storage) -> Vec { + s.get_due_periodic(now_millis()) + .unwrap() + .into_iter() + .map(|p| p.name) + .collect() +} + +fn test_periodic_crud(s: &impl Storage) { + use taskito_core::storage::models::NewPeriodicTaskRow; + let past = now_millis() - 1_000; + let row = |name: &'static str| NewPeriodicTaskRow { + name, + task_name: "periodic-task", + cron_expr: "* * * * *", + args: None, + kwargs: None, + queue: "default", + enabled: true, + next_run: past, + timezone: None, + }; + s.register_periodic(&row("pc-a")).unwrap(); + s.register_periodic(&row("pc-b")).unwrap(); + + // list_periodic returns every registered task. + let listed: Vec = s + .list_periodic() + .unwrap() + .into_iter() + .map(|p| p.name) + .collect(); + assert!(listed.contains(&"pc-a".to_string()) && listed.contains(&"pc-b".to_string())); + + // Pausing drops it from the due set but keeps it in the catalog. + assert!(s.set_periodic_enabled("pc-a", false).unwrap()); + assert!(!due_periodic_names(s).contains(&"pc-a".to_string())); + assert!(s.list_periodic().unwrap().iter().any(|p| p.name == "pc-a")); + + // Resuming makes it due again. + assert!(s.set_periodic_enabled("pc-a", true).unwrap()); + assert!(due_periodic_names(s).contains(&"pc-a".to_string())); + + // Toggling or deleting an unknown task reports "not found". + assert!(!s.set_periodic_enabled("pc-missing", true).unwrap()); + + assert!(s.delete_periodic("pc-a").unwrap()); + assert!(!s.list_periodic().unwrap().iter().any(|p| p.name == "pc-a")); + assert!(!s.delete_periodic("pc-a").unwrap()); +} + fn run_storage_tests(s: &impl Storage) { test_enqueue_and_get(s); test_dequeue(s); @@ -647,6 +698,7 @@ fn run_storage_tests(s: &impl Storage) { test_record_and_get_errors(s); test_workers(s); test_pause_resume_queue(s); + test_periodic_crud(s); test_circuit_breakers(s); test_execution_claims_purge(s); test_dashboard_settings(s); From 80a842fae37091c6885df6acc039479bad70647c Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 27 Jun 2026 13:06:26 +0530 Subject: [PATCH 2/5] feat(python): expose periodic list/delete/pause/resume Native PyQueue bindings over the new storage ops, plus type stubs. --- crates/taskito-python/src/py_queue/mod.rs | 33 +++++++++++++++++++++++ sdks/python/taskito/_taskito.pyi | 4 +++ 2 files changed, 37 insertions(+) diff --git a/crates/taskito-python/src/py_queue/mod.rs b/crates/taskito-python/src/py_queue/mod.rs index 557bb3a6..a107c6b5 100644 --- a/crates/taskito-python/src/py_queue/mod.rs +++ b/crates/taskito-python/src/py_queue/mod.rs @@ -664,6 +664,39 @@ impl PyQueue { .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string())) } + /// List all registered periodic tasks as `(name, task_name, cron_expr, enabled)` tuples. + pub fn list_periodic(&self) -> PyResult> { + let rows = self + .storage + .list_periodic() + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?; + Ok(rows + .into_iter() + .map(|row| (row.name, row.task_name, row.cron_expr, row.enabled)) + .collect()) + } + + /// Remove a periodic task. Returns false if no task had that name. + pub fn delete_periodic(&self, name: &str) -> PyResult { + self.storage + .delete_periodic(name) + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string())) + } + + /// Pause a periodic task so it stops firing. Returns false if no task had that name. + pub fn pause_periodic(&self, name: &str) -> PyResult { + self.storage + .set_periodic_enabled(name, false) + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string())) + } + + /// Resume a paused periodic task. Returns false if no task had that name. + pub fn resume_periodic(&self, name: &str) -> PyResult { + self.storage + .set_periodic_enabled(name, true) + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string())) + } + pub fn __repr__(&self) -> String { format!( "PyQueue(db_path='{}', workers={})", diff --git a/sdks/python/taskito/_taskito.pyi b/sdks/python/taskito/_taskito.pyi index ce26f4c8..ea43dfc4 100644 --- a/sdks/python/taskito/_taskito.pyi +++ b/sdks/python/taskito/_taskito.pyi @@ -177,6 +177,10 @@ class PyQueue: queue: str = "default", timezone: str | None = None, ) -> None: ... + def list_periodic(self) -> list[tuple[str, str, str, bool]]: ... + def delete_periodic(self, name: str) -> bool: ... + def pause_periodic(self, name: str) -> bool: ... + def resume_periodic(self, name: str) -> bool: ... def run_worker( self, task_registry: dict[str, Any], From 2785dcca038955740767b63f5373b3afe0c15c91 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 27 Jun 2026 13:06:28 +0530 Subject: [PATCH 3/5] feat(java): periodic CRUD (list/delete/pause/resume) Queue.listPeriodic/deletePeriodic/pausePeriodic/resumePeriodic over the new core ops, with a PeriodicInfo view and an in-memory backend impl. Live-verified by PeriodicTest. --- crates/taskito-java/src/convert.rs | 31 +++++++++- crates/taskito-java/src/queue/periodic.rs | 62 +++++++++++++++++-- .../org/byteveda/taskito/DefaultQueue.java | 21 +++++++ .../main/java/org/byteveda/taskito/Queue.java | 13 ++++ .../taskito/internal/JniQueueBackend.java | 15 +++++ .../taskito/internal/NativeQueue.java | 9 +++ .../byteveda/taskito/model/PeriodicInfo.java | 41 ++++++++++++ .../byteveda/taskito/spi/QueueBackend.java | 15 +++++ .../org/byteveda/taskito/PeriodicTest.java | 58 +++++++++++++++++ .../taskito/test/InMemoryQueueBackend.java | 34 +++++++++- 10 files changed, 292 insertions(+), 7 deletions(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/model/PeriodicInfo.java create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/PeriodicTest.java diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs index f7c3b08b..53e7de8d 100644 --- a/crates/taskito-java/src/convert.rs +++ b/crates/taskito-java/src/convert.rs @@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize}; use taskito_core::job::{now_millis, Job, NewJob}; use taskito_core::storage::models::{ - JobErrorRow, LockInfoRow, TaskLogRow, TaskMetricRow, WorkerRow, + JobErrorRow, LockInfoRow, PeriodicTaskRow, TaskLogRow, TaskMetricRow, WorkerRow, }; use taskito_core::storage::{DeadJob, QueueStats}; @@ -203,6 +203,35 @@ pub fn status_code(status: &str) -> Option { } } +/// Periodic-task view (omits the opaque args/kwargs payloads). +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct PeriodicTaskView<'a> { + pub name: &'a str, + pub task_name: &'a str, + pub cron_expr: &'a str, + pub queue: &'a str, + pub enabled: bool, + pub last_run: Option, + pub next_run: i64, + pub timezone: Option<&'a str>, +} + +impl<'a> From<&'a PeriodicTaskRow> for PeriodicTaskView<'a> { + fn from(p: &'a PeriodicTaskRow) -> Self { + Self { + name: &p.name, + task_name: &p.task_name, + cron_expr: &p.cron_expr, + queue: &p.queue, + enabled: p.enabled, + last_run: p.last_run, + next_run: p.next_run, + timezone: p.timezone.as_deref(), + } + } +} + /// Dead-letter entry view (omits the opaque payload). #[derive(Serialize)] #[serde(rename_all = "camelCase")] diff --git a/crates/taskito-java/src/queue/periodic.rs b/crates/taskito-java/src/queue/periodic.rs index 42035e0d..2ed8c874 100644 --- a/crates/taskito-java/src/queue/periodic.rs +++ b/crates/taskito-java/src/queue/periodic.rs @@ -1,16 +1,18 @@ -//! Periodic (cron) task registration. The worker maintenance loop enqueues due -//! tasks; this binds only registration. Re-registering a name replaces it. +//! Periodic (cron) task registration + management. The worker maintenance loop +//! enqueues due tasks; this binds registration plus list / delete / pause / +//! resume. Re-registering a name replaces it. use jni::objects::{JByteArray, JClass, JString}; -use jni::sys::{jboolean, jlong}; +use jni::sys::{jboolean, jlong, jstring, JNI_FALSE}; use jni::JNIEnv; use taskito_core::job::now_millis; use taskito_core::periodic::{next_cron_time, next_cron_time_tz}; use taskito_core::storage::models::NewPeriodicTaskRow; use taskito_core::Storage; -use super::borrow_queue; -use crate::ffi::{guard, read_bytes, read_optional_string, read_string}; +use super::{borrow_queue, to_jboolean}; +use crate::convert::{to_json, PeriodicTaskView}; +use crate::ffi::{guard, new_string, read_bytes, read_optional_string, read_string}; const DEFAULT_QUEUE: &str = "default"; @@ -64,3 +66,53 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_registerPe Ok(next_run) }) } + +/// `String listPeriodic(long handle)` — a JSON array of registered periodic +/// tasks (enabled and paused). +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listPeriodic<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let tasks = queue.storage.list_periodic()?; + let views: Vec = tasks.iter().map(PeriodicTaskView::from).collect(); + new_string(env, to_json(&views)?) + }) +} + +/// `boolean deletePeriodic(long handle, String name)` — false if none had that name. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_deletePeriodic<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + name: JString<'local>, +) -> jboolean { + guard(&mut env, JNI_FALSE, |env| { + let queue = unsafe { borrow_queue(handle) }; + let name = read_string(env, &name)?; + Ok(to_jboolean(queue.storage.delete_periodic(&name)?)) + }) +} + +/// `boolean setPeriodicEnabled(long handle, String name, boolean enabled)` — +/// pause (false) or resume (true); false if none had that name. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_setPeriodicEnabled<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + name: JString<'local>, + enabled: jboolean, +) -> jboolean { + guard(&mut env, JNI_FALSE, |env| { + let queue = unsafe { borrow_queue(handle) }; + let name = read_string(env, &name)?; + Ok(to_jboolean( + queue.storage.set_periodic_enabled(&name, enabled != 0)?, + )) + }) +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java index 03548bb5..285c42f5 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java @@ -21,6 +21,7 @@ import org.byteveda.taskito.model.Job; import org.byteveda.taskito.model.JobError; import org.byteveda.taskito.model.JobFilter; +import org.byteveda.taskito.model.PeriodicInfo; import org.byteveda.taskito.model.QueueStats; import org.byteveda.taskito.model.TaskLog; import org.byteveda.taskito.model.TaskMetric; @@ -315,6 +316,26 @@ public long registerPeriodic(PeriodicTask task) { task.name, task.taskName, task.cron, payload, task.queue, task.timezone, task.enabled); } + @Override + public List listPeriodic() { + return decodeList(backend.listPeriodicJson(), PeriodicInfo.class); + } + + @Override + public boolean deletePeriodic(String name) { + return backend.deletePeriodic(name); + } + + @Override + public boolean pausePeriodic(String name) { + return backend.setPeriodicEnabled(name, false); + } + + @Override + public boolean resumePeriodic(String name) { + return backend.setPeriodicEnabled(name, true); + } + // ── Workflows ─────────────────────────────────────────────────── @Override diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java index 64db53fa..c05e26ad 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java @@ -11,6 +11,7 @@ import org.byteveda.taskito.model.Job; import org.byteveda.taskito.model.JobError; import org.byteveda.taskito.model.JobFilter; +import org.byteveda.taskito.model.PeriodicInfo; import org.byteveda.taskito.model.QueueStats; import org.byteveda.taskito.model.TaskLog; import org.byteveda.taskito.model.TaskMetric; @@ -141,6 +142,18 @@ public interface Queue extends AutoCloseable { /** Register (or replace) a cron task; returns the next fire time (Unix ms). */ long registerPeriodic(PeriodicTask task); + /** Every registered periodic task, enabled or paused. */ + List listPeriodic(); + + /** Unschedule a periodic task; false if none had that name. */ + boolean deletePeriodic(String name); + + /** Stop a periodic task from firing without removing it; false if none had that name. */ + boolean pausePeriodic(String name); + + /** Resume a paused periodic task; false if none had that name. */ + boolean resumePeriodic(String name); + // ── Workflows ─────────────────────────────────────────────────── /** Submit a workflow DAG; returns a handle to the run. */ diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java index c59c64e6..077ef919 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java @@ -190,6 +190,21 @@ public long registerPeriodic( return NativeQueue.registerPeriodic(handle, name, taskName, cron, args, queue, timezone, enabled); } + @Override + public String listPeriodicJson() { + return NativeQueue.listPeriodic(handle); + } + + @Override + public boolean deletePeriodic(String name) { + return NativeQueue.deletePeriodic(handle, name); + } + + @Override + public boolean setPeriodicEnabled(String name, boolean enabled) { + return NativeQueue.setPeriodicEnabled(handle, name, enabled); + } + @Override public String submitWorkflow( String name, diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java index fca76a3b..39843c8d 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeQueue.java @@ -107,6 +107,15 @@ public static native long registerPeriodic( String timezone, boolean enabled); + /** A JSON array of every registered periodic task (enabled and paused). */ + public static native String listPeriodic(long handle); + + /** Remove a periodic task; false if none had that name. */ + public static native boolean deletePeriodic(long handle, String name); + + /** Pause (false) or resume (true) a periodic task; false if none had that name. */ + public static native boolean setPeriodicEnabled(long handle, String name, boolean enabled); + // ── Worker ────────────────────────────────────────────────────── /** Start a worker; returns its handle. {@code bridge} is a {@code WorkerBridge}. */ public static native long runWorker(long handle, Object bridge, String optionsJson); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/PeriodicInfo.java b/sdks/java/src/main/java/org/byteveda/taskito/model/PeriodicInfo.java new file mode 100644 index 00000000..b6b697cd --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/PeriodicInfo.java @@ -0,0 +1,41 @@ +package org.byteveda.taskito.model; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** Immutable view of a registered periodic task. Timestamps are Unix milliseconds. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class PeriodicInfo { + public final String name; + public final String taskName; + public final String cronExpr; + public final String queue; + public final boolean enabled; + /** Last fire time, or {@code null} if it has not run yet. */ + public final Long lastRun; + + public final long nextRun; + /** IANA timezone the cron is evaluated in, or {@code null} for UTC. */ + public final String timezone; + + @JsonCreator + public PeriodicInfo( + @JsonProperty("name") String name, + @JsonProperty("taskName") String taskName, + @JsonProperty("cronExpr") String cronExpr, + @JsonProperty("queue") String queue, + @JsonProperty("enabled") boolean enabled, + @JsonProperty("lastRun") Long lastRun, + @JsonProperty("nextRun") long nextRun, + @JsonProperty("timezone") String timezone) { + this.name = name; + this.taskName = taskName; + this.cronExpr = cronExpr; + this.queue = queue; + this.enabled = enabled; + this.lastRun = lastRun; + this.nextRun = nextRun; + this.timezone = timezone; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java index b0766eda..80151ebb 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java @@ -107,6 +107,21 @@ default long registerPeriodic( throw new UnsupportedOperationException("periodic tasks not supported by this backend"); } + /** A JSON array of every registered periodic task. */ + default String listPeriodicJson() { + throw new UnsupportedOperationException("periodic tasks not supported by this backend"); + } + + /** Remove a periodic task; false if none had that name. */ + default boolean deletePeriodic(String name) { + throw new UnsupportedOperationException("periodic tasks not supported by this backend"); + } + + /** Pause (false) or resume (true) a periodic task; false if none had that name. */ + default boolean setPeriodicEnabled(String name, boolean enabled) { + throw new UnsupportedOperationException("periodic tasks not supported by this backend"); + } + // ── Workflows ─────────────────────────────────────────────────── // Optional capability: default to throwing so existing custom backends keep // compiling and fail explicitly only when workflows are actually used. diff --git a/sdks/java/src/test/java/org/byteveda/taskito/PeriodicTest.java b/sdks/java/src/test/java/org/byteveda/taskito/PeriodicTest.java new file mode 100644 index 00000000..ebf1c111 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/PeriodicTest.java @@ -0,0 +1,58 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.util.List; +import java.util.Optional; +import org.byteveda.taskito.model.PeriodicInfo; +import org.byteveda.taskito.scheduling.PeriodicTask; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** Periodic CRUD over the JNI backend: list / pause-resume (toggles enabled) / delete. */ +class PeriodicTest { + + private static PeriodicInfo find(List tasks, String name) { + Optional match = + tasks.stream().filter(p -> p.name.equals(name)).findFirst(); + assertTrue(match.isPresent(), "expected periodic task '" + name + "'"); + return match.get(); + } + + @Test + @Timeout(30) + void listPauseResumeDelete(@TempDir Path dir) { + try (Queue queue = Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open()) { + queue.registerPeriodic( + PeriodicTask.builder("nightly", "report", "0 0 0 * * *").build()); + queue.registerPeriodic( + PeriodicTask.builder("hourly", "sync", "0 0 * * * *").build()); + + assertEquals(2, queue.listPeriodic().size()); + assertTrue(find(queue.listPeriodic(), "nightly").enabled); + + // Pause toggles enabled but keeps the task in the catalog. + assertTrue(queue.pausePeriodic("nightly")); + assertFalse(find(queue.listPeriodic(), "nightly").enabled); + assertEquals(2, queue.listPeriodic().size()); + + assertTrue(queue.resumePeriodic("nightly")); + assertTrue(find(queue.listPeriodic(), "nightly").enabled); + + // Unknown name → not found. + assertFalse(queue.pausePeriodic("ghost")); + + // Delete removes it; a second delete reports not-found. + assertTrue(queue.deletePeriodic("nightly")); + assertEquals(1, queue.listPeriodic().size()); + assertFalse(queue.deletePeriodic("nightly")); + } + } +} diff --git a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java index c3f9ef18..4358659c 100644 --- a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java +++ b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java @@ -30,6 +30,7 @@ public final class InMemoryQueueBackend implements QueueBackend { private final Map jobs = new ConcurrentHashMap<>(); private final Map settings = new ConcurrentHashMap<>(); + private final Map> periodics = new ConcurrentHashMap<>(); private final Map locks = new ConcurrentHashMap<>(); private final List> dead = new CopyOnWriteArrayList<>(); private final Map>> logs = new ConcurrentHashMap<>(); @@ -363,14 +364,45 @@ public Optional lockInfoJson(String name) { } // ── Periodic ──────────────────────────────────────────────────── + // Catalog only — the in-memory backend records periodic tasks but never + // fires them. list/delete/pause exercise the same management surface as JNI. @Override public long registerPeriodic( String name, String taskName, String cron, byte[] args, String queue, String timezone, boolean enabled) { - settings.put("periodic:" + name, taskName + "|" + cron); // recorded, never fires + Map rec = new LinkedHashMap<>(); + rec.put("name", name); + rec.put("taskName", taskName); + rec.put("cronExpr", cron); + rec.put("queue", queue); + rec.put("enabled", enabled); + rec.put("lastRun", null); + rec.put("nextRun", now()); + rec.put("timezone", timezone); + periodics.put(name, rec); return now(); } + @Override + public String listPeriodicJson() { + return toJson(new ArrayList<>(periodics.values())); + } + + @Override + public boolean deletePeriodic(String name) { + return periodics.remove(name) != null; + } + + @Override + public boolean setPeriodicEnabled(String name, boolean enabled) { + Map rec = periodics.get(name); + if (rec == null) { + return false; + } + rec.put("enabled", enabled); + return true; + } + // ── Worker ────────────────────────────────────────────────────── @Override From 70ff483e086e5742b95289bd6dc31f55762fbfa5 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 27 Jun 2026 13:15:48 +0530 Subject: [PATCH 4/5] feat(node): periodic list/delete/pause/resume Queue.listPeriodic/deletePeriodic/pausePeriodic/resumePeriodic over the new core ops, with a JsPeriodicTask napi view. Live-verified by the periodic test. --- crates/taskito-node/src/convert/mod.rs | 2 ++ crates/taskito-node/src/convert/periodic.rs | 32 +++++++++++++++++++++ crates/taskito-node/src/queue/periodic.rs | 29 +++++++++++++++++-- sdks/node/src/index.ts | 1 + sdks/node/src/queue.ts | 21 ++++++++++++++ sdks/node/src/types.ts | 14 +++++++++ sdks/node/test/worker/periodic.test.ts | 31 ++++++++++++++++++++ 7 files changed, 127 insertions(+), 3 deletions(-) create mode 100644 crates/taskito-node/src/convert/periodic.rs diff --git a/crates/taskito-node/src/convert/mod.rs b/crates/taskito-node/src/convert/mod.rs index 69ae8e97..44433cd6 100644 --- a/crates/taskito-node/src/convert/mod.rs +++ b/crates/taskito-node/src/convert/mod.rs @@ -5,6 +5,7 @@ mod job; mod lock; mod log; mod outcome; +mod periodic; mod stats; mod task_config; #[cfg(feature = "workflows")] @@ -14,6 +15,7 @@ pub use job::{build_new_job, job_to_js, JsJob, JsTaskInvocation}; pub use lock::{lock_info_to_js, JsLockInfo}; pub use log::{log_to_js, JsTaskLog}; pub use outcome::{outcome_to_js, JsOutcome}; +pub use periodic::{periodic_to_js, JsPeriodicTask}; pub use stats::{ dead_job_to_js, job_error_to_js, metric_to_js, stats_to_js, status_code, worker_to_js, JsDeadJob, JsJobError, JsMetric, JsStats, JsWorkerRow, diff --git a/crates/taskito-node/src/convert/periodic.rs b/crates/taskito-node/src/convert/periodic.rs new file mode 100644 index 00000000..df4fa460 --- /dev/null +++ b/crates/taskito-node/src/convert/periodic.rs @@ -0,0 +1,32 @@ +//! JS-facing shape for a registered periodic task. Timestamps are Unix milliseconds. + +use napi_derive::napi; +use taskito_core::storage::models::PeriodicTaskRow; + +/// JS-facing view of a periodic task (omits the opaque args/kwargs payloads). +#[napi(object)] +pub struct JsPeriodicTask { + pub name: String, + pub task_name: String, + pub cron_expr: String, + pub queue: String, + pub enabled: bool, + /// Last fire time (Unix ms), or `null` if it has not run yet. + pub last_run: Option, + pub next_run: i64, + /// IANA timezone the cron is evaluated in, or `null` for UTC. + pub timezone: Option, +} + +pub fn periodic_to_js(row: PeriodicTaskRow) -> JsPeriodicTask { + JsPeriodicTask { + name: row.name, + task_name: row.task_name, + cron_expr: row.cron_expr, + queue: row.queue, + enabled: row.enabled, + last_run: row.last_run, + next_run: row.next_run, + timezone: row.timezone, + } +} diff --git a/crates/taskito-node/src/queue/periodic.rs b/crates/taskito-node/src/queue/periodic.rs index 13991559..609bae3a 100644 --- a/crates/taskito-node/src/queue/periodic.rs +++ b/crates/taskito-node/src/queue/periodic.rs @@ -1,6 +1,6 @@ -//! Periodic (cron) task registration. The worker's maintenance loop enqueues -//! due tasks; this binds only registration. Re-registering an existing name -//! replaces it (upsert). +//! Periodic (cron) task registration + management. The worker's maintenance +//! loop enqueues due tasks; this binds registration plus list / delete / pause +//! / resume. Re-registering an existing name replaces it (upsert). use napi::bindgen_prelude::{Buffer, Result}; use napi_derive::napi; @@ -10,6 +10,7 @@ use taskito_core::storage::models::NewPeriodicTaskRow; use taskito_core::Storage; use super::JsQueue; +use crate::convert::{periodic_to_js, JsPeriodicTask}; use crate::error::to_napi_err; const DEFAULT_QUEUE: &str = "default"; @@ -54,4 +55,26 @@ impl JsQueue { self.storage.register_periodic(&row).map_err(to_napi_err)?; Ok(next_run) } + + /// Every registered periodic task, enabled or paused. + #[napi] + pub fn list_periodic(&self) -> Result> { + let tasks = self.storage.list_periodic().map_err(to_napi_err)?; + Ok(tasks.into_iter().map(periodic_to_js).collect()) + } + + /// Unschedule a periodic task. Returns false if none had that name. + #[napi] + pub fn delete_periodic(&self, name: String) -> Result { + self.storage.delete_periodic(&name).map_err(to_napi_err) + } + + /// Pause (false) or resume (true) a periodic task by toggling its enabled + /// flag. Returns false if none had that name. + #[napi] + pub fn set_periodic_enabled(&self, name: String, enabled: bool) -> Result { + self.storage + .set_periodic_enabled(&name, enabled) + .map_err(to_napi_err) + } } diff --git a/sdks/node/src/index.ts b/sdks/node/src/index.ts index 65ec9ce2..e0378d3f 100644 --- a/sdks/node/src/index.ts +++ b/sdks/node/src/index.ts @@ -57,6 +57,7 @@ export type { MeshWorkerConfig, Metric, PeriodicOptions, + PeriodicTask, QueueLimits, RateLimit, ResultOptions, diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index 1aad96a0..f5f51eac 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -37,6 +37,7 @@ import type { JobFilter, Metric, PeriodicOptions, + PeriodicTask, QueueLimits, RegisteredTask, ResultOptions, @@ -161,6 +162,26 @@ export class Queue { ); } + /** Every registered periodic task, enabled or paused. */ + listPeriodic(): PeriodicTask[] { + return this.native.listPeriodic(); + } + + /** Unschedule a periodic task. Returns false if none had that name. */ + deletePeriodic(name: string): boolean { + return this.native.deletePeriodic(name); + } + + /** Stop a periodic task from firing without removing it; false if none had that name. */ + pausePeriodic(name: string): boolean { + return this.native.setPeriodicEnabled(name, false); + } + + /** Resume a paused periodic task; false if none had that name. */ + resumePeriodic(name: string): boolean { + return this.native.setPeriodicEnabled(name, true); + } + /** * Register a task that receives injected resources as a trailing `deps` object * (`handler(...args, deps)`). The `deps` param is stripped from the typed diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts index a1887fd4..4a6a0438 100644 --- a/sdks/node/src/types.ts +++ b/sdks/node/src/types.ts @@ -95,6 +95,20 @@ export interface PeriodicOptions { enabled?: boolean; } +/** A registered periodic task, as returned by {@link Queue.listPeriodic}. Timestamps are Unix ms. */ +export interface PeriodicTask { + name: string; + taskName: string; + cronExpr: string; + queue: string; + enabled: boolean; + /** Last fire time, or absent if it has not run yet. */ + lastRun?: number; + nextRun: number; + /** IANA timezone the cron is evaluated in, or absent for UTC. */ + timezone?: string; +} + /** Per-queue resilience config. */ export interface QueueLimits { maxConcurrent?: number; diff --git a/sdks/node/test/worker/periodic.test.ts b/sdks/node/test/worker/periodic.test.ts index 3ba0476f..515ae9f4 100644 --- a/sdks/node/test/worker/periodic.test.ts +++ b/sdks/node/test/worker/periodic.test.ts @@ -37,6 +37,37 @@ it("rejects an invalid cron expression", () => { expect(() => queue.registerPeriodic("bad", "beat", "not a cron")).toThrow(); }); +it("lists, pauses, resumes, and deletes periodic tasks", () => { + const queue = freshQueue(); + queue.task("beat", () => null); + queue.registerPeriodic("nightly", "beat", "0 0 0 * * *"); + queue.registerPeriodic("hourly", "beat", "0 0 * * * *"); + + expect( + queue + .listPeriodic() + .map((p) => p.name) + .sort(), + ).toEqual(["hourly", "nightly"]); + expect(queue.listPeriodic().find((p) => p.name === "nightly")?.enabled).toBe(true); + + // Pause toggles `enabled` but keeps the task in the catalog. + expect(queue.pausePeriodic("nightly")).toBe(true); + expect(queue.listPeriodic().find((p) => p.name === "nightly")?.enabled).toBe(false); + expect(queue.listPeriodic()).toHaveLength(2); + + expect(queue.resumePeriodic("nightly")).toBe(true); + expect(queue.listPeriodic().find((p) => p.name === "nightly")?.enabled).toBe(true); + + // Unknown name → not found. + expect(queue.pausePeriodic("ghost")).toBe(false); + + // Delete removes it; a second delete reports not-found. + expect(queue.deletePeriodic("nightly")).toBe(true); + expect(queue.listPeriodic()).toHaveLength(1); + expect(queue.deletePeriodic("nightly")).toBe(false); +}); + it("accepts a circuit-breaker config and still runs the task", async () => { const queue = freshQueue(); queue.task("guarded", (x: number) => x * 2, { From 168cfdee186e596780948a497694a85de35bfc0e Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Sat, 27 Jun 2026 13:53:10 +0530 Subject: [PATCH 5/5] fix(core): list Redis periodic tasks via SCAN, drop index MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The periodic:all index only saw tasks registered after this change, hiding pre-existing Redis periodics from list_periodic. Scan the periodic:* keyspace instead (as list_dead/list_workers do) — no secondary index to keep consistent, and tasks from any version are listed. --- .../src/storage/redis_backend/periodic.rs | 49 ++++++++++++------- 1 file changed, 31 insertions(+), 18 deletions(-) diff --git a/crates/taskito-core/src/storage/redis_backend/periodic.rs b/crates/taskito-core/src/storage/redis_backend/periodic.rs index 299d97f4..7b3e1bd2 100644 --- a/crates/taskito-core/src/storage/redis_backend/periodic.rs +++ b/crates/taskito-core/src/storage/redis_backend/periodic.rs @@ -57,11 +57,9 @@ impl RedisStorage { let pkey = self.key(&["periodic", task.name]); let due_key = self.key(&["periodic", "due"]); - let all_key = self.key(&["periodic", "all"]); let pipe = &mut redis::pipe(); pipe.set(&pkey, &json); - pipe.sadd(&all_key, task.name); if entry.enabled { pipe.zadd(&due_key, task.name, task.next_run as f64); } @@ -118,27 +116,44 @@ impl RedisStorage { Ok(()) } - /// List all registered periodic tasks, enabled or paused. + /// List all registered periodic tasks, enabled or paused. Scans the + /// `periodic:*` keyspace (like `list_dead`/`list_workers`) so there is no + /// secondary index to keep consistent — tasks registered by any version are + /// listed. pub fn list_periodic(&self) -> Result> { let mut conn = self.conn()?; - let all_key = self.key(&["periodic", "all"]); - - let names: Vec = conn.smembers(&all_key).map_err(map_err)?; + let pattern = self.key(&["periodic", "*"]); + // The due sorted-set shares the `periodic:` namespace but is not a task + // JSON blob, so skip it (a `GET` on it would be a WRONGTYPE error). + let due_key = self.key(&["periodic", "due"]); let mut rows = Vec::new(); - for name in names { - let pkey = self.key(&["periodic", &name]); - let data: Option = conn.get(&pkey).map_err(map_err)?; - match data { - Some(d) => { + let mut cursor: u64 = 0; + loop { + let (next_cursor, keys): (u64, Vec) = redis::cmd("SCAN") + .arg(cursor) + .arg("MATCH") + .arg(&pattern) + .arg("COUNT") + .arg(100) + .query(&mut conn) + .map_err(map_err)?; + + for key in keys { + if key == due_key { + continue; + } + let data: Option = conn.get(&key).map_err(map_err)?; + if let Some(d) = data { let entry: PeriodicEntry = serde_json::from_str(&d).map_err(|e| QueueError::Other(e.to_string()))?; rows.push(PeriodicTaskRow::from(entry)); } - // Stale index entry: the task key is gone, so self-heal. - None => { - let _removed: i64 = conn.srem(&all_key, &name).map_err(map_err)?; - } + } + + cursor = next_cursor; + if cursor == 0 { + break; } } @@ -150,13 +165,11 @@ impl RedisStorage { let mut conn = self.conn()?; let pkey = self.key(&["periodic", name]); let due_key = self.key(&["periodic", "due"]); - let all_key = self.key(&["periodic", "all"]); let pipe = &mut redis::pipe(); pipe.del(&pkey); pipe.zrem(&due_key, name); - pipe.srem(&all_key, name); - let (deleted, _zrem, _srem): (i64, i64, i64) = pipe.query(&mut conn).map_err(map_err)?; + let (deleted, _zrem): (i64, i64) = pipe.query(&mut conn).map_err(map_err)?; Ok(deleted > 0) }