diff --git a/crates/taskito-core/src/storage/cursor.rs b/crates/taskito-core/src/storage/cursor.rs index 7ccf1781..79da1d7a 100644 --- a/crates/taskito-core/src/storage/cursor.rs +++ b/crates/taskito-core/src/storage/cursor.rs @@ -31,10 +31,48 @@ pub fn decode_cursor(cursor: &str) -> Result<(i64, &str)> { Ok((sort_key, id)) } +/// Build the next-page cursor from a returned page, or `None` when the page is +/// the last one — a short page means the listing is exhausted, so callers can +/// loop `while let Some(cursor)`. `key` reads the row's `(sort_key, id)`. +/// +/// Lives here rather than in a binding: every SDK pages the same way, and the +/// "fewer rows than limit ⇒ last page" rule is part of the cursor contract, not +/// of any one shell. +pub fn next_cursor(rows: &[T], limit: i64, key: impl Fn(&T) -> (i64, &str)) -> Option { + if (rows.len() as i64) < limit { + return None; + } + rows.last().map(|row| { + let (sort_key, id) = key(row); + encode_cursor(sort_key, id) + }) +} + #[cfg(test)] mod tests { use super::*; + /// `(sort_key, id)` rows standing in for a listing's page. + fn rows(n: usize) -> Vec<(i64, String)> { + (0..n).map(|i| (i as i64, format!("id-{i}"))).collect() + } + + #[test] + fn next_cursor_is_none_on_a_short_page() { + assert!(next_cursor(&rows(2), 5, |r| (r.0, &r.1)).is_none()); + } + + #[test] + fn next_cursor_points_at_the_last_row_of_a_full_page() { + let cursor = next_cursor(&rows(3), 3, |r| (r.0, &r.1)).expect("full page has a cursor"); + assert_eq!(cursor, encode_cursor(2, "id-2")); + } + + #[test] + fn next_cursor_is_none_on_an_empty_page() { + assert!(next_cursor::<(i64, String)>(&[], 5, |r| (r.0, &r.1)).is_none()); + } + #[test] fn round_trips() { let s = encode_cursor(1_720_000_000_123, "0190a1b2-c3d4-7e5f-8a9b-0c1d2e3f4a5b"); diff --git a/crates/taskito-java/src/convert.rs b/crates/taskito-java/src/convert.rs index f0c7416d..c2ea2557 100644 --- a/crates/taskito-java/src/convert.rs +++ b/crates/taskito-java/src/convert.rs @@ -359,6 +359,15 @@ pub struct JobFilter { pub offset: Option, } +/// One page of a keyset-paginated listing, as the Java `Page` decodes it. +/// `next_cursor` is absent on the last page. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct PageView { + pub items: Vec, + pub next_cursor: Option, +} + /// Map a lowercase status string to the core's `i32` status code. pub fn status_code(status: &str) -> Option { match status { diff --git a/crates/taskito-java/src/queue/inspect.rs b/crates/taskito-java/src/queue/inspect.rs index 5c958693..df73d3a5 100644 --- a/crates/taskito-java/src/queue/inspect.rs +++ b/crates/taskito-java/src/queue/inspect.rs @@ -12,7 +12,7 @@ use taskito_core::Storage; use super::borrow_queue; use crate::convert::{ status_code, to_json, CircuitBreakerView, JobErrorView, JobFilter, JobView, MetricView, - StatsView, WorkerView, + PageView, StatsView, WorkerView, }; use crate::error::BindingError; use crate::ffi::{guard, new_string, read_optional_string, read_string}; @@ -116,6 +116,86 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listJobs<' }) } +/// `String listJobsAfter(long handle, String filterJson, String afterOrNull)` — +/// keyset-paginated `listJobs`, ordered by created time. Returns a `Page`; its +/// `nextCursor` is absent on the last page. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listJobsAfter<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + filter_json: JString<'local>, + after: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let raw = read_string(env, &filter_json)?; + let filter: JobFilter = crate::convert::parse_json(&raw, "job filter")?; + // Reject an unknown status so a typo fails loudly instead of matching all. + let status = match filter.status.as_deref() { + Some(s) => Some( + status_code(s).ok_or_else(|| BindingError::new(format!("unknown status '{s}'")))?, + ), + None => None, + }; + let limit = filter.limit.unwrap_or(DEFAULT_LIMIT).max(0); + // A malformed cursor is a bad request, not a reason to silently restart + // from the first page. + let after = read_optional_string(env, &after)?; + let cursor = after + .as_deref() + .map(taskito_core::storage::cursor::decode_cursor) + .transpose()?; + let jobs = queue.storage.list_jobs_after( + status, + filter.queue.as_deref(), + filter.task.as_deref(), + limit, + cursor, + queue.namespace.as_deref(), + )?; + let next_cursor = + taskito_core::storage::cursor::next_cursor(&jobs, limit, |j| (j.created_at, &j.id)); + let page = PageView { + items: jobs.iter().map(JobView::from).collect(), + next_cursor, + }; + new_string(env, to_json(&page)?) + }) +} + +/// `String listArchivedAfter(long handle, long limit, String afterOrNull)` — +/// keyset-paginated `listArchived`, ordered by completed time. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_listArchivedAfter<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + limit: jlong, + after: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let limit = limit.max(0); + let after = read_optional_string(env, &after)?; + let cursor = after + .as_deref() + .map(taskito_core::storage::cursor::decode_cursor) + .transpose()?; + let jobs = queue.storage.list_archived_after(limit, cursor)?; + // Archived rows always carry `completed_at`; the fallback keeps the key + // non-null for a row written before it was set. + let next_cursor = taskito_core::storage::cursor::next_cursor(&jobs, limit, |j| { + (j.completed_at.unwrap_or(j.created_at), &j.id) + }); + let page = PageView { + items: jobs.iter().map(JobView::from).collect(), + next_cursor, + }; + new_string(env, to_json(&page)?) + }) +} + /// `String jobErrors(long handle, String jobId)` — one entry per failed attempt. #[no_mangle] pub extern "system" fn Java_org_byteveda_taskito_internal_NativeQueue_jobErrors<'local>( diff --git a/crates/taskito-node/src/config.rs b/crates/taskito-node/src/config.rs index 02806571..be7dfcf9 100644 --- a/crates/taskito-node/src/config.rs +++ b/crates/taskito-node/src/config.rs @@ -88,6 +88,28 @@ pub struct JobFilter { pub offset: Option, } +/// Filter for [`crate::queue::JsQueue::list_jobs_filtered`]: everything +/// [`JobFilter`] matches on, plus substring and created-at-range predicates. +/// All fields optional. +#[napi(object)] +#[derive(Default)] +pub struct DetailedJobFilter { + /// Lowercase status (`"pending"`, `"running"`, `"complete"`, `"failed"`, `"dead"`, `"cancelled"`). + pub status: Option, + pub queue: Option, + pub task: Option, + /// Substring matched against the job's metadata. + pub metadata_like: Option, + /// Substring matched against the job's error. + pub error_like: Option, + /// Only jobs created at or after this Unix-millisecond timestamp. + pub created_after: Option, + /// Only jobs created at or before this Unix-millisecond timestamp. + pub created_before: Option, + pub limit: Option, + pub offset: Option, +} + /// Circuit-breaker config for a task: trip after `threshold` failures within /// `windowMs`, stay open for `cooldownMs`, then probe before fully closing. #[napi(object)] diff --git a/crates/taskito-node/src/convert/job.rs b/crates/taskito-node/src/convert/job.rs index fd215da4..0e4fc55e 100644 --- a/crates/taskito-node/src/convert/job.rs +++ b/crates/taskito-node/src/convert/job.rs @@ -115,3 +115,15 @@ pub fn job_to_js(job: Job) -> JsJob { namespace: job.namespace, } } + +/// One page of jobs plus the cursor for the next. napi has no generics, so each +/// item type needs its own page struct; the TS shell declares a `Page` they +/// all satisfy structurally. +/// +/// `next_cursor` is `null` on the last page — a short page means the listing is +/// exhausted — so callers can loop until it is null. +#[napi(object)] +pub struct JsJobPage { + pub items: Vec, + pub next_cursor: Option, +} diff --git a/crates/taskito-node/src/convert/mod.rs b/crates/taskito-node/src/convert/mod.rs index 16b9b06b..af0d7b33 100644 --- a/crates/taskito-node/src/convert/mod.rs +++ b/crates/taskito-node/src/convert/mod.rs @@ -13,7 +13,7 @@ mod task_config; #[cfg(feature = "workflows")] mod workflow; -pub use job::{build_new_job, job_to_js, JsJob, JsTaskInvocation}; +pub use job::{build_new_job, job_to_js, JsJob, JsJobPage, JsTaskInvocation}; pub(crate) use job::{DEFAULT_MAX_RETRIES, DEFAULT_PRIORITY, DEFAULT_TIMEOUT_MS}; pub use lock::{lock_info_to_js, JsLockInfo}; pub use log::{log_to_js, JsTaskLog}; @@ -25,7 +25,7 @@ pub use periodic::{periodic_to_js, JsPeriodicTask}; pub use pubsub::{subscription_to_js, JsSubscription}; 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, + JsDeadJob, JsDeadJobPage, JsJobError, JsMetric, JsStats, JsWorkerRow, }; pub use task_config::{queue_config, task_config}; #[cfg(feature = "workflows")] diff --git a/crates/taskito-node/src/convert/stats.rs b/crates/taskito-node/src/convert/stats.rs index b1959798..23b0f718 100644 --- a/crates/taskito-node/src/convert/stats.rs +++ b/crates/taskito-node/src/convert/stats.rs @@ -145,3 +145,11 @@ pub fn status_code(status: &str) -> Option { _ => None, } } + +/// One page of dead-letter entries plus the cursor for the next. See +/// [`crate::convert::JsJobPage`] for the contract. +#[napi(object)] +pub struct JsDeadJobPage { + pub items: Vec, + pub next_cursor: Option, +} diff --git a/crates/taskito-node/src/queue/admin.rs b/crates/taskito-node/src/queue/admin.rs index af8b22eb..a402d083 100644 --- a/crates/taskito-node/src/queue/admin.rs +++ b/crates/taskito-node/src/queue/admin.rs @@ -10,7 +10,7 @@ use taskito_core::job::{now_millis, NewJob}; use taskito_core::Storage; use super::JsQueue; -use crate::convert::{dead_job_to_js, JsDeadJob}; +use crate::convert::{dead_job_to_js, JsDeadJob, JsDeadJobPage}; use crate::error::{invalid_arg, join_to_napi_err, non_negative, to_napi_err}; const DEFAULT_LIMIT: i64 = 50; @@ -35,6 +35,38 @@ impl JsQueue { .map_err(join_to_napi_err)? } + /// Keyset-paginated [`JsQueue::dead_letters`], ordered by failed time. Pass + /// a page's `nextCursor` back as `after`; `null` means the last page. + #[napi] + pub async fn dead_letters_after( + &self, + limit: Option, + after: Option, + ) -> Result { + let limit = non_negative(limit.unwrap_or(DEFAULT_LIMIT), "limit")?; + let storage = self.storage.clone(); + spawn_blocking(move || { + // A malformed cursor is a bad request, not a reason to silently + // restart from the first page. + let cursor = after + .as_deref() + .map(taskito_core::storage::cursor::decode_cursor) + .transpose() + .map_err(to_napi_err)?; + let dead = storage + .list_dead_after(limit, cursor) + .map_err(to_napi_err)?; + let next_cursor = + taskito_core::storage::cursor::next_cursor(&dead, limit, |d| (d.failed_at, &d.id)); + Ok(JsDeadJobPage { + items: dead.into_iter().map(dead_job_to_js).collect(), + next_cursor, + }) + }) + .await + .map_err(join_to_napi_err)? + } + /// List dead-letter entries for a single task (paginated, newest first). #[napi] pub async fn dead_letters_by_task( diff --git a/crates/taskito-node/src/queue/inspect.rs b/crates/taskito-node/src/queue/inspect.rs index e1c96429..c571bbc4 100644 --- a/crates/taskito-node/src/queue/inspect.rs +++ b/crates/taskito-node/src/queue/inspect.rs @@ -9,16 +9,66 @@ use napi_derive::napi; use taskito_core::Storage; use super::JsQueue; -use crate::config::JobFilter; +use crate::config::{DetailedJobFilter, JobFilter}; use crate::convert::{ circuit_breaker_to_js, job_error_to_js, job_to_js, log_to_js, metric_to_js, replay_to_js, stats_to_js, status_code, worker_to_js, JsCircuitBreaker, JsDagEdge, JsJob, JsJobDag, - JsJobError, JsMetric, JsReplayEntry, JsStats, JsTaskLog, JsWorkerRow, + JsJobError, JsJobPage, JsMetric, JsReplayEntry, JsStats, JsTaskLog, JsWorkerRow, }; use crate::error::{invalid_arg, join_to_napi_err, non_negative, to_napi_err}; const DEFAULT_LIMIT: i64 = 50; +/// Resolve a lowercase status name to its code. An unrecognized status would +/// otherwise silently widen the result set to every job, so a typo fails loudly. +fn parse_status_filter(status: Option<&str>) -> Result> { + match status { + Some(s) => status_code(s) + .map(Some) + .ok_or_else(|| invalid_arg(format!("unknown status filter '{s}'"))), + None => Ok(None), + } +} + +/// Reject an offset on a keyset call. The two ways of paging do not compose — +/// a cursor already says where to resume — so honouring one and ignoring the +/// other would quietly return a different page than the caller asked for. +fn reject_offset(offset: Option) -> Result<()> { + match offset { + Some(o) if o != 0 => Err(invalid_arg( + "offset is not supported with cursor pagination; pass the page's nextCursor as `after`", + )), + _ => Ok(()), + } +} + +/// Decode a caller-supplied page cursor. A malformed one is a bad request, not +/// a reason to silently restart from the first page. +fn parse_cursor(after: Option<&str>) -> Result> { + after + .map(taskito_core::storage::cursor::decode_cursor) + .transpose() + .map_err(to_napi_err) +} + +/// Wrap a page of jobs with the cursor for the next, keyed on `sort_key`. +/// Archived rows always carry `completed_at`; the fallback keeps the key +/// non-null for a row written before it was set. +fn job_page(jobs: Vec, limit: i64, by_completed_at: bool) -> JsJobPage { + let next_cursor = taskito_core::storage::cursor::next_cursor(&jobs, limit, |j| { + let key = if by_completed_at { + j.completed_at.unwrap_or(j.created_at) + } else { + j.created_at + }; + (key, &j.id) + }); + JsJobPage { + items: jobs.into_iter().map(job_to_js).collect(), + next_cursor, + } +} + #[napi] impl JsQueue { /// Job counts by status across all queues. @@ -60,15 +110,7 @@ impl JsQueue { #[napi] pub async fn list_jobs(&self, filter: Option) -> Result> { let filter = filter.unwrap_or_default(); - // An unrecognized status would otherwise silently widen the result set - // to every job; reject it so a typo fails loudly. - let status = match filter.status.as_deref() { - Some(s) => Some( - status_code(s) - .ok_or_else(|| invalid_arg(format!("unknown status filter '{s}'")))?, - ), - None => None, - }; + let status = parse_status_filter(filter.status.as_deref())?; let limit = non_negative(filter.limit.unwrap_or(DEFAULT_LIMIT), "limit")?; let offset = non_negative(filter.offset.unwrap_or(0), "offset")?; let storage = self.storage.clone(); @@ -90,6 +132,151 @@ impl JsQueue { .map_err(join_to_napi_err)? } + /// List jobs on the wider filter: everything [`JsQueue::list_jobs`] matches + /// on, plus metadata/error substrings and a created-at range. + #[napi] + pub async fn list_jobs_filtered( + &self, + filter: Option, + ) -> Result> { + let filter = filter.unwrap_or_default(); + let status = parse_status_filter(filter.status.as_deref())?; + let limit = non_negative(filter.limit.unwrap_or(DEFAULT_LIMIT), "limit")?; + let offset = non_negative(filter.offset.unwrap_or(0), "offset")?; + let storage = self.storage.clone(); + let namespace = self.namespace.clone(); + spawn_blocking(move || { + let jobs = storage + .list_jobs_filtered( + status, + filter.queue.as_deref(), + filter.task.as_deref(), + filter.metadata_like.as_deref(), + filter.error_like.as_deref(), + filter.created_after, + filter.created_before, + limit, + offset, + namespace.as_deref(), + ) + .map_err(to_napi_err)?; + Ok(jobs.into_iter().map(job_to_js).collect()) + }) + .await + .map_err(join_to_napi_err)? + } + + /// List archived (completed, moved out of the live table) jobs, newest first. + #[napi] + pub async fn list_archived( + &self, + limit: Option, + offset: Option, + ) -> Result> { + let limit = non_negative(limit.unwrap_or(DEFAULT_LIMIT), "limit")?; + let offset = non_negative(offset.unwrap_or(0), "offset")?; + let storage = self.storage.clone(); + spawn_blocking(move || { + let jobs = storage.list_archived(limit, offset).map_err(to_napi_err)?; + Ok(jobs.into_iter().map(job_to_js).collect()) + }) + .await + .map_err(join_to_napi_err)? + } + + /// Keyset-paginated [`JsQueue::list_jobs`], ordered by created time. Pass a + /// page's `nextCursor` back as `after`; `null` means the last page. + /// + /// O(page) at any depth on SQLite/Postgres. On Redis the status indexes are + /// not seekable, so the keyset is applied in memory — correct, but O(rows + /// matching the filter) rather than O(page). + #[napi] + pub async fn list_jobs_after( + &self, + filter: Option, + after: Option, + ) -> Result { + let filter = filter.unwrap_or_default(); + let status = parse_status_filter(filter.status.as_deref())?; + reject_offset(filter.offset)?; + let limit = non_negative(filter.limit.unwrap_or(DEFAULT_LIMIT), "limit")?; + let storage = self.storage.clone(); + let namespace = self.namespace.clone(); + spawn_blocking(move || { + let cursor = parse_cursor(after.as_deref())?; + let jobs = storage + .list_jobs_after( + status, + filter.queue.as_deref(), + filter.task.as_deref(), + limit, + cursor, + namespace.as_deref(), + ) + .map_err(to_napi_err)?; + Ok(job_page(jobs, limit, false)) + }) + .await + .map_err(join_to_napi_err)? + } + + /// Keyset-paginated [`JsQueue::list_jobs_filtered`], ordered by created + /// time. See [`JsQueue::list_jobs_after`] for the cursor contract. + #[napi] + pub async fn list_jobs_filtered_after( + &self, + filter: Option, + after: Option, + ) -> Result { + let filter = filter.unwrap_or_default(); + let status = parse_status_filter(filter.status.as_deref())?; + reject_offset(filter.offset)?; + let limit = non_negative(filter.limit.unwrap_or(DEFAULT_LIMIT), "limit")?; + let storage = self.storage.clone(); + let namespace = self.namespace.clone(); + spawn_blocking(move || { + let cursor = parse_cursor(after.as_deref())?; + let jobs = storage + .list_jobs_filtered_after( + status, + filter.queue.as_deref(), + filter.task.as_deref(), + filter.metadata_like.as_deref(), + filter.error_like.as_deref(), + filter.created_after, + filter.created_before, + limit, + cursor, + namespace.as_deref(), + ) + .map_err(to_napi_err)?; + Ok(job_page(jobs, limit, false)) + }) + .await + .map_err(join_to_napi_err)? + } + + /// Keyset-paginated [`JsQueue::list_archived`], ordered by completed time. + /// See [`JsQueue::list_jobs_after`] for the cursor contract. + #[napi] + pub async fn list_archived_after( + &self, + limit: Option, + after: Option, + ) -> Result { + let limit = non_negative(limit.unwrap_or(DEFAULT_LIMIT), "limit")?; + let storage = self.storage.clone(); + spawn_blocking(move || { + let cursor = parse_cursor(after.as_deref())?; + let jobs = storage + .list_archived_after(limit, cursor) + .map_err(to_napi_err)?; + Ok(job_page(jobs, limit, true)) + }) + .await + .map_err(join_to_napi_err)? + } + /// Error history for a job (one entry per failed attempt). #[napi] pub async fn get_job_errors(&self, job_id: String) -> Result> { diff --git a/crates/taskito-python/src/py_queue/mod.rs b/crates/taskito-python/src/py_queue/mod.rs index 7d3c2bcc..ea9ba2b3 100644 --- a/crates/taskito-python/src/py_queue/mod.rs +++ b/crates/taskito-python/src/py_queue/mod.rs @@ -26,22 +26,7 @@ use taskito_core::worker::WorkerDispatcher; use crate::py_job::PyJob; -/// Build the next-page cursor from a returned page. `None` when the page is the -/// last one (fewer rows than `limit`), so callers can loop `while cursor`. -/// Shared by every `*_after` binding across the `py_queue` submodules. -pub(crate) fn next_cursor( - rows: &[T], - limit: i64, - key: impl Fn(&T) -> (i64, &str), -) -> Option { - if (rows.len() as i64) < limit { - return None; - } - rows.last().map(|r| { - let (sort_key, id) = key(r); - taskito_core::storage::cursor::encode_cursor(sort_key, id) - }) -} +pub(crate) use taskito_core::storage::cursor::next_cursor; /// The core queue engine exposed to Python. #[pyclass] diff --git a/sdks/java/graalvm-smoke/src/main/java/org/byteveda/taskito/graalvm/Smoke.java b/sdks/java/graalvm-smoke/src/main/java/org/byteveda/taskito/graalvm/Smoke.java index e47969f2..39d212c2 100644 --- a/sdks/java/graalvm-smoke/src/main/java/org/byteveda/taskito/graalvm/Smoke.java +++ b/sdks/java/graalvm-smoke/src/main/java/org/byteveda/taskito/graalvm/Smoke.java @@ -7,6 +7,7 @@ import java.util.concurrent.TimeUnit; import org.byteveda.taskito.Taskito; import org.byteveda.taskito.events.EventName; +import org.byteveda.taskito.model.JobFilter; import org.byteveda.taskito.scheduling.PeriodicTask; import org.byteveda.taskito.task.Task; import org.byteveda.taskito.worker.Worker; @@ -47,13 +48,18 @@ public static void main(String[] args) throws Exception { throw new IllegalStateException("unexpected result: " + value); } - // Exercise the Jackson DTO paths (PeriodicInfo + DeadJob deserialization). + // Exercise the Jackson DTO paths (PeriodicInfo + DeadJob + Page + // deserialization). The agent generates the reflection metadata from + // what this actually calls, so a DTO the smoke never touches gets no + // entry and fails under --no-fallback at runtime. queue.registerPeriodic( PeriodicTask.builder("nightly", "echo", "0 0 0 * * *").build()); if (queue.listPeriodic().isEmpty()) { throw new IllegalStateException("listPeriodic returned empty"); } queue.listDead(10, 0); + // Page is generic, so its element type has to be exercised too. + queue.listJobsAfter(JobFilter.builder().limit(10).build(), null); System.out.println("taskito graalvm smoke ok"); } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java index a3bfdb55..32bb1441 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -26,6 +26,7 @@ import org.byteveda.taskito.interception.Interception; import org.byteveda.taskito.interception.Interceptor; import org.byteveda.taskito.internal.IdempotencyKeys; +import org.byteveda.taskito.internal.MiddlewareDisables; import org.byteveda.taskito.locks.Lock; import org.byteveda.taskito.locks.LockInfo; import org.byteveda.taskito.middleware.EnqueueContext; @@ -36,6 +37,7 @@ import org.byteveda.taskito.model.JobDag; import org.byteveda.taskito.model.JobError; import org.byteveda.taskito.model.JobFilter; +import org.byteveda.taskito.model.Page; import org.byteveda.taskito.model.PeriodicInfo; import org.byteveda.taskito.model.QueueStats; import org.byteveda.taskito.model.ReplayEntry; @@ -468,6 +470,16 @@ public List listJobs(JobFilter filter) { return decodeList(backend.listJobsJson(encode(filter)), Job.class); } + @Override + public Page listJobsAfter(JobFilter filter, String after) { + return decodePage(backend.listJobsAfterJson(encode(filter), after), Job.class); + } + + @Override + public Page listArchivedAfter(long limit, String after) { + return decodePage(backend.listArchivedAfterJson(limit, after), Job.class); + } + @Override public List jobErrors(String jobId) { return decodeList(backend.jobErrorsJson(jobId), JobError.class); @@ -1055,6 +1067,37 @@ private static R decode(String json, Class type) { } } + @Override + public void disableMiddleware(String taskName, String middlewareName) { + List disabled = new ArrayList<>(listDisabledMiddleware(taskName)); + if (!disabled.contains(middlewareName)) { + disabled.add(middlewareName); + backend.setSetting(MiddlewareDisables.key(taskName), encode(disabled)); + } + } + + @Override + public void enableMiddleware(String taskName, String middlewareName) { + List disabled = new ArrayList<>(listDisabledMiddleware(taskName)); + if (disabled.remove(middlewareName)) { + backend.setSetting(MiddlewareDisables.key(taskName), encode(disabled)); + } + } + + @Override + public List listDisabledMiddleware(String taskName) { + return new MiddlewareDisables(backend).disabledFor(taskName); + } + + private static Page decodePage(String json, Class element) { + JavaType type = VIEWS.getTypeFactory().constructParametricType(Page.class, element); + try { + return VIEWS.readValue(json, type); + } catch (Exception e) { + throw new SerializationException("failed to decode native response", e); + } + } + private static List decodeList(String json, Class element) { JavaType type = VIEWS.getTypeFactory().constructCollectionType(List.class, element); try { diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java index 00873c3e..803ff455 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java @@ -28,6 +28,7 @@ import org.byteveda.taskito.model.JobDag; import org.byteveda.taskito.model.JobError; import org.byteveda.taskito.model.JobFilter; +import org.byteveda.taskito.model.Page; import org.byteveda.taskito.model.PeriodicInfo; import org.byteveda.taskito.model.QueueStats; import org.byteveda.taskito.model.ReplayEntry; @@ -179,6 +180,29 @@ static Builder builder() { List listJobs(JobFilter filter); + /** + * Keyset-paginated {@link #listJobs}, ordered by created time. Pass a page's + * {@code nextCursor} back as {@code after}; {@code null} starts at the first + * page, and a {@code null} {@code nextCursor} means the last one. + * + *

Stays O(page) at any depth, unlike an offset walk. On Redis the status + * indexes are not seekable, so the keyset is applied in memory — correct, but + * O(matching rows) rather than O(page). + * + * @param filter same predicates as {@link #listJobs}; its {@code offset} is ignored + * @param after cursor from a previous page, or {@code null} for the first + */ + Page listJobsAfter(JobFilter filter, String after); + + /** + * Keyset-paginated archived-job listing, ordered by completed time. See + * {@link #listJobsAfter} for the cursor contract. + * + * @param limit page size + * @param after cursor from a previous page, or {@code null} for the first + */ + Page listArchivedAfter(long limit, String after); + List jobErrors(String jobId); /** Per-execution metrics within the last {@code sinceMs}; null task = all. */ @@ -231,6 +255,35 @@ static Builder builder() { Map listSettings(); + // ── Middleware toggles ────────────────────────────────────────── + + /** + * Stop running {@code middlewareName} for {@code taskName}. Takes effect on + * the next job — workers read the list per invocation, so no restart is + * needed. The name is the middleware's fully-qualified class name. + * + * @param taskName task to disable it for + * @param middlewareName fully-qualified class name of the middleware + */ + void disableMiddleware(String taskName, String middlewareName); + + /** + * Undo {@link #disableMiddleware}. A no-op when it was not disabled. + * + * @param taskName task to re-enable it for + * @param middlewareName fully-qualified class name of the middleware + */ + void enableMiddleware(String taskName, String middlewareName); + + /** + * Middleware names currently disabled for {@code taskName}; empty when none + * are. + * + * @param taskName task to read the list for + * @return the disabled middleware names + */ + List listDisabledMiddleware(String taskName); + // ── Logs ──────────────────────────────────────────────────────── void writeTaskLog(String jobId, String taskName, String level, String message); 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 afdb67fe..a1172755 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 @@ -107,6 +107,16 @@ public String listJobsJson(String filterJson) { return withOpenHandle(() -> NativeQueue.listJobs(handle, filterJson)); } + @Override + public String listJobsAfterJson(String filterJson, String afterOrNull) { + return withOpenHandle(() -> NativeQueue.listJobsAfter(handle, filterJson, afterOrNull)); + } + + @Override + public String listArchivedAfterJson(long limit, String afterOrNull) { + return withOpenHandle(() -> NativeQueue.listArchivedAfter(handle, limit, afterOrNull)); + } + @Override public String jobErrorsJson(String jobId) { return withOpenHandle(() -> NativeQueue.jobErrors(handle, jobId)); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/MiddlewareDisables.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/MiddlewareDisables.java new file mode 100644 index 00000000..d217e629 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/MiddlewareDisables.java @@ -0,0 +1,84 @@ +package org.byteveda.taskito.internal; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.List; +import java.util.Optional; +import org.byteveda.taskito.logging.TaskitoLogger; +import org.byteveda.taskito.middleware.Middleware; +import org.byteveda.taskito.spi.QueueBackend; + +/** + * Per-task middleware disable list, persisted under + * {@code middleware:disabled:} as a JSON array of middleware names. + * + *

Read on every invocation rather than cached, so a toggle takes effect on + * the next job without restarting the worker. That costs one settings read per + * job — the same trade the other SDKs accept for a live toggle. + * + *

Internal: the key format and the name derivation live here so the worker + * that honours the list and the admin API that writes it cannot drift apart. + */ +public final class MiddlewareDisables { + private static final TaskitoLogger LOG = TaskitoLogger.create("worker"); + private static final ObjectMapper JSON = new ObjectMapper(); + private static final TypeReference> NAMES = new TypeReference>() {}; + + private static final String KEY_PREFIX = "middleware:disabled:"; + + private final QueueBackend backend; + + public MiddlewareDisables(QueueBackend backend) { + this.backend = backend; + } + + /** The settings key holding {@code taskName}'s disable list. */ + public static String key(String taskName) { + return KEY_PREFIX + taskName; + } + + /** + * The stable name a middleware is keyed on. Java has no name property to + * prefer, so the class name is the only identity available — the same one + * already used when logging a middleware failure. + */ + public static String nameOf(Middleware middleware) { + return middleware.getClass().getName(); + } + + /** Names disabled for {@code taskName}; empty when none are, or the list is unreadable. */ + public List disabledFor(String taskName) { + Optional raw = backend.getSetting(key(taskName)); + if (raw.isEmpty()) { + return List.of(); + } + try { + List names = JSON.readValue(raw.get(), NAMES); + return names == null ? List.of() : names; + } catch (Exception e) { + // A corrupt list must not stop the job — run every middleware, which + // is the same behaviour as having no list at all. + LOG.warn("middleware disable list for " + taskName + " is not valid JSON; treating as empty"); + return List.of(); + } + } + + /** + * {@code middleware} minus anything disabled for {@code taskName}. + * + *

Callers must resolve this once per invocation and reuse it for + * {@code before}/{@code after}/{@code onError}: re-reading between them would + * let a mid-job toggle run {@code after} on a middleware whose {@code before} + * never ran, an unbalanced pair authors are entitled to assume cannot happen. + */ + public List resolve(String taskName, List middleware) { + if (middleware.isEmpty()) { + return middleware; + } + List disabled = disabledFor(taskName); + if (disabled.isEmpty()) { + return middleware; + } + return middleware.stream().filter(m -> !disabled.contains(nameOf(m))).toList(); + } +} 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 97c1eccc..46e3bb7f 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 @@ -47,6 +47,10 @@ private NativeQueue() {} public static native String listJobs(long handle, String filterJson); + public static native String listJobsAfter(long handle, String filterJson, String afterOrNull); + + public static native String listArchivedAfter(long handle, long limit, String afterOrNull); + public static native String jobErrors(long handle, String jobId); public static native String metrics(long handle, String taskNameOrNull, long sinceMs); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/Page.java b/sdks/java/src/main/java/org/byteveda/taskito/model/Page.java new file mode 100644 index 00000000..309d8f90 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/Page.java @@ -0,0 +1,43 @@ +package org.byteveda.taskito.model; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import java.util.List; + +/** + * One page of a keyset-paginated listing. + * + *

{@code nextCursor} is {@code null} on the last page — a page shorter than + * the requested limit means the listing is exhausted — so a walk reads: + * + * 

{@code
+ * JobFilter filter = JobFilter.builder().limit(50).build();
+ * String cursor = null;
+ * do {
+ *     Page page = queue.listJobsAfter(filter, cursor);
+ *     page.items.forEach(this::handle);
+ *     cursor = page.nextCursor;
+ * } while (cursor != null);
+ * }
+ * + *

Pass the cursor back verbatim and treat it as opaque; its format is an + * implementation detail and may change. + * + * @param the listing's element type + */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class Page { + public final List items; + public final String nextCursor; + + /** + * @param items this page's rows, newest first + * @param nextCursor cursor for the next page, or {@code null} on the last one + */ + @JsonCreator + public Page(@JsonProperty("items") List items, @JsonProperty("nextCursor") String nextCursor) { + this.items = items == null ? List.of() : List.copyOf(items); + this.nextCursor = nextCursor; + } +} 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 b4bc52f7..688569d4 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 @@ -46,6 +46,20 @@ public interface QueueBackend extends AutoCloseable { String listJobsJson(String filterJson); + /** + * A keyset-paginated page of jobs as JSON. Defaults to unsupported: a backend + * without a seekable ordering cannot honour the cursor contract, and silently + * returning an unpaginated list would look like the last page. + */ + default String listJobsAfterJson(String filterJson, String afterOrNull) { + throw new UnsupportedOperationException("keyset pagination not supported by this backend"); + } + + /** A keyset-paginated page of archived jobs as JSON. See {@link #listJobsAfterJson}. */ + default String listArchivedAfterJson(long limit, String afterOrNull) { + throw new UnsupportedOperationException("keyset pagination not supported by this backend"); + } + String jobErrorsJson(String jobId); String metricsJson(String taskNameOrNull, long sinceMs); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java index b64528de..114def0e 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java @@ -12,6 +12,7 @@ import org.byteveda.taskito.events.Emitter; import org.byteveda.taskito.events.EventName; import org.byteveda.taskito.events.OutcomeEvent; +import org.byteveda.taskito.internal.MiddlewareDisables; import org.byteveda.taskito.logging.TaskitoLogger; import org.byteveda.taskito.middleware.JobInfo; import org.byteveda.taskito.middleware.Middleware; @@ -44,6 +45,7 @@ final class WorkerDispatchBridge implements WorkerBridge { private final List middleware; private final ResourceRuntime resources; private final Map codecs; + private final MiddlewareDisables disables; // Resolved once startWorker returns; job tasks await it before completing. private final CompletableFuture control = new CompletableFuture<>(); @@ -64,6 +66,7 @@ final class WorkerDispatchBridge implements WorkerBridge { this.middleware = middleware; this.resources = resources; this.codecs = codecs; + this.disables = new MiddlewareDisables(backend); } void bind(WorkerControl control) { @@ -90,18 +93,27 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { if (scope != null) { Resources.enter(scope); } + // Empty until resolved, so a failure to read the disable list runs onError + // on nothing — which is right, because no before() ran either. + List chain = List.of(); try { - for (Middleware m : middleware) { + // Resolved once and reused below: re-reading the disable list between + // before and after would let a mid-job toggle run after on a middleware + // whose before never ran. Inside the try because it reads the backend, + // so a settings failure fails the job rather than leaving it unresolved + // with its resource scope still bound. + chain = disables.resolve(taskName, middleware); + for (Middleware m : chain) { m.before(context); } Object argument = serializer.deserializeCall(decodePayload(payload, task.codecs), task.payloadType); Object result = task.handler.apply(argument); - for (Middleware m : middleware) { + for (Middleware m : chain) { m.after(context, result); } bound.completeJob(token, serializer.serialize(result)); } catch (Throwable t) { - for (Middleware m : middleware) { + for (Middleware m : chain) { m.onError(context, t); } // Canonical structured error (middleware above saw the live Throwable). @@ -132,7 +144,7 @@ public void onOutcome(String kind, String jobId, String taskName, String error, EventName name = EventName.fromKind(kind); OutcomeEvent event = new OutcomeEvent(name, jobId, taskName, error, retryCount, timedOut); emitter.emit(event); - for (Middleware m : middleware) { + for (Middleware m : disables.resolve(taskName, middleware)) { try { dispatch(m, name, event); } catch (RuntimeException e) { diff --git a/sdks/java/src/test/java/org/byteveda/taskito/core/PaginationTest.java b/sdks/java/src/test/java/org/byteveda/taskito/core/PaginationTest.java new file mode 100644 index 00000000..44468e40 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/core/PaginationTest.java @@ -0,0 +1,96 @@ +package org.byteveda.taskito.core; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.function.Function; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.model.Job; +import org.byteveda.taskito.model.JobFilter; +import org.byteveda.taskito.model.Page; +import org.byteveda.taskito.task.Task; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** Keyset pagination walks a listing exactly once and terminates. */ +class PaginationTest { + + /** Walk every page, collecting ids, so a dropped or repeated row shows up. */ + private static List walk(Function> fetch) { + List ids = new ArrayList<>(); + String cursor = null; + int pages = 0; + do { + Page page = fetch.apply(cursor); + page.items.forEach(job -> ids.add(job.id)); + cursor = page.nextCursor; + if (++pages > 20) { + throw new IllegalStateException("pagination did not terminate"); + } + } while (cursor != null); + return ids; + } + + @Test + @Timeout(30) + void walksEveryJobExactlyOnce(@TempDir Path dir) { + Task paged = Task.of("paged", String.class); + + try (Taskito queue = Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open()) { + Set enqueued = new HashSet<>(); + for (int i = 0; i < 5; i++) { + enqueued.add(queue.enqueue(paged, String.valueOf(i))); + } + + List seen = walk( + cursor -> queue.listJobsAfter(JobFilter.builder().limit(2).build(), cursor)); + + assertEquals(5, seen.size(), "every job appears exactly once"); + assertEquals(enqueued, new HashSet<>(seen), "the walk covers exactly what was enqueued"); + } + } + + @Test + @Timeout(30) + void aShortPageEndsTheWalk(@TempDir Path dir) { + Task paged = Task.of("paged", String.class); + + try (Taskito queue = Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open()) { + queue.enqueue(paged, "1"); + + Page page = queue.listJobsAfter(JobFilter.builder().limit(10).build(), null); + assertEquals(1, page.items.size()); + assertNull(page.nextCursor, "a page shorter than the limit is the last one"); + } + } + + @Test + @Timeout(30) + void malformedCursorIsRejected(@TempDir Path dir) { + try (Taskito queue = Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open()) { + RuntimeException error = assertThrows( + RuntimeException.class, + () -> queue.listJobsAfter(JobFilter.builder().limit(2).build(), "nocolon")); + assertTrue( + error.getMessage().contains("cursor"), + "a bad cursor must be rejected, not silently restarted: " + error.getMessage()); + } + } +} diff --git a/sdks/java/src/test/java/org/byteveda/taskito/worker/MiddlewareDisableTest.java b/sdks/java/src/test/java/org/byteveda/taskito/worker/MiddlewareDisableTest.java new file mode 100644 index 00000000..b94882d0 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/worker/MiddlewareDisableTest.java @@ -0,0 +1,129 @@ +package org.byteveda.taskito.worker; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.events.EventName; +import org.byteveda.taskito.middleware.Middleware; +import org.byteveda.taskito.middleware.TaskContext; +import org.byteveda.taskito.task.Task; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** Per-task middleware disables take effect without restarting the worker. */ +class MiddlewareDisableTest { + + /** Counts how many times its before/after ran, so a skip is visible. */ + static final class Counting implements Middleware { + final AtomicInteger before = new AtomicInteger(); + final AtomicInteger after = new AtomicInteger(); + + @Override + public void before(TaskContext context) { + before.incrementAndGet(); + } + + @Override + public void after(TaskContext context, Object result) { + after.incrementAndGet(); + } + } + + /** Poll the aggregate rather than rendezvous the running worker on a barrier. */ + private static void awaitCompleted(Taskito queue, int expected) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(20); + while (queue.stats().completed < expected && System.nanoTime() < deadline) { + Thread.sleep(20); + } + assertEquals(expected, queue.stats().completed, "expected " + expected + " completed jobs"); + } + + @Test + @Timeout(30) + void disableTakesEffectOnTheNextJobWithoutARestart(@TempDir Path dir) throws Exception { + Task counted = Task.of("counted", String.class); + Counting counting = new Counting(); + + try (Taskito queue = Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open()) { + queue.use(counting); + + CountDownLatch first = new CountDownLatch(1); + try (Worker worker = queue.worker() + .handle(counted, (String p) -> p) + .on(EventName.SUCCESS, event -> first.countDown()) + .start()) { + queue.enqueue(counted, "1"); + assertTrue(first.await(20, TimeUnit.SECONDS), "first job should run"); + assertEquals(1, counting.before.get(), "middleware runs while enabled"); + + // Toggle mid-worker: the list is read per invocation, so the next + // job must skip it without the worker being restarted. + queue.disableMiddleware("counted", Counting.class.getName()); + assertEquals(List.of(Counting.class.getName()), queue.listDisabledMiddleware("counted")); + + queue.enqueue(counted, "2"); + awaitCompleted(queue, 2); + assertEquals(1, counting.before.get(), "disabled middleware must not run"); + + // Re-enabling restores it, again with no restart. + queue.enableMiddleware("counted", Counting.class.getName()); + assertEquals(List.of(), queue.listDisabledMiddleware("counted")); + queue.enqueue(counted, "3"); + awaitCompleted(queue, 3); + assertEquals(2, counting.before.get(), "re-enabled middleware runs again"); + } + } + } + + @Test + @Timeout(30) + void beforeAndAfterShareOneResolvedChain(@TempDir Path dir) throws Exception { + // The chain is resolved once per invocation, so a toggle landing while a + // job is in flight cannot run after on a middleware whose before did not. + Task slow = Task.of("slow", String.class); + Counting counting = new Counting(); + + try (Taskito queue = Taskito.builder() + .backend("sqlite") + .url(dir.resolve("t.db").toString()) + .open()) { + queue.use(counting); + + CountDownLatch started = new CountDownLatch(1); + try (Worker worker = queue.worker() + .handle(slow, (String p) -> { + started.countDown(); + try { + Thread.sleep(400); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return p; + }) + .start()) { + queue.enqueue(slow, "1"); + assertTrue(started.await(20, TimeUnit.SECONDS), "handler should start"); + + // Disable while the job is mid-flight. + queue.disableMiddleware("slow", Counting.class.getName()); + + awaitCompleted(queue, 1); + } + } + + assertEquals( + counting.before.get(), + counting.after.get(), + "before and after must stay balanced across a mid-job toggle"); + } +} 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 254fd383..ed673c77 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 @@ -215,6 +215,87 @@ public String listJobsJson(String filterJson) { return toJson(views.subList(from, to)); } + @Override + public String listJobsAfterJson(String filterJson, String afterOrNull) { + JsonNode filter = readNode(filterJson); + String status = text(filter, "status"); + String queue = text(filter, "queue"); + String task = text(filter, "task"); + int limit = Math.max(0, optInt(filter, "limit", Integer.MAX_VALUE)); + + List matching = new ArrayList<>(); + for (JobRec job : jobs.values()) { + if (status != null && !status.equals(job.status)) { + continue; + } + if (queue != null && !queue.equals(job.queue)) { + continue; + } + if (task != null && !task.equals(job.taskName)) { + continue; + } + matching.add(job); + } + return keysetPage(matching, limit, afterOrNull, job -> job.createdAt); + } + + @Override + public String listArchivedAfterJson(long limit, String afterOrNull) { + List archived = new ArrayList<>(); + for (JobRec job : jobs.values()) { + if (job.completedAt != null) { + archived.add(job); + } + } + // Clamp before narrowing: the native handlers reject a negative limit, and + // subList would throw on one rather than return an empty page. + long clamped = Math.max(0, limit); + return keysetPage( + archived, + (int) Math.min(clamped, Integer.MAX_VALUE), + afterOrNull, + job -> job.completedAt == null ? job.createdAt : job.completedAt); + } + + /** + * Order by {@code (sortKey, id)} descending, seek past the cursor, take a + * page — the same contract the native backends implement, so a test against + * this backend exercises the real pagination semantics rather than a stub. + */ + private static String keysetPage( + List rows, int limit, String afterOrNull, java.util.function.ToLongFunction sortKey) { + rows.sort((a, b) -> { + int byKey = Long.compare(sortKey.applyAsLong(b), sortKey.applyAsLong(a)); + return byKey != 0 ? byKey : b.id.compareTo(a.id); + }); + if (afterOrNull != null) { + int split = afterOrNull.indexOf(':'); + if (split < 0) { + throw new IllegalArgumentException("invalid cursor: " + afterOrNull); + } + long cursorKey = Long.parseLong(afterOrNull.substring(0, split)); + String cursorId = afterOrNull.substring(split + 1); + rows.removeIf(job -> { + long key = sortKey.applyAsLong(job); + return key > cursorKey || (key == cursorKey && job.id.compareTo(cursorId) >= 0); + }); + } + List page = rows.subList(0, Math.min(limit, rows.size())); + List> views = new ArrayList<>(); + for (JobRec job : page) { + views.add(jobView(job)); + } + Map result = new LinkedHashMap<>(); + result.put("items", views); + // A short page is the last one. + result.put( + "nextCursor", + page.size() < limit || page.isEmpty() + ? null + : sortKey.applyAsLong(page.get(page.size() - 1)) + ":" + page.get(page.size() - 1).id); + return toJson(result); + } + @Override public String jobErrorsJson(String jobId) { return "[]"; diff --git a/sdks/node/src/index.ts b/sdks/node/src/index.ts index 6cfc2096..a5a59b8e 100644 --- a/sdks/node/src/index.ts +++ b/sdks/node/src/index.ts @@ -79,13 +79,17 @@ export { decodeTaskError, encodeTaskError, type TaskError } from "./task-error"; export type { AnyHandler, CircuitBreakerOptions, + CursorDetailedJobFilter, + CursorJobFilter, DeadJob, + DetailedJobFilter, EnqueueOptions, Job, JobError, JobFilter, MeshWorkerConfig, Metric, + Page, PeriodicOptions, PeriodicTask, PublishOptions, diff --git a/sdks/node/src/native.ts b/sdks/node/src/native.ts index 063065a3..444c3702 100644 --- a/sdks/node/src/native.ts +++ b/sdks/node/src/native.ts @@ -17,6 +17,7 @@ export type NativeWorker = InstanceType; export type { CircuitBreakerInput, + DetailedJobFilter, EnqueueOptions, JobFilter, JsCircuitBreaker, diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index 5deeca86..a59abc3d 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -50,13 +50,17 @@ import { import type { AnyHandler, CircuitBreaker, + CursorDetailedJobFilter, + CursorJobFilter, DeadJob, + DetailedJobFilter, EnqueueOptions, Job, JobDag, JobError, JobFilter, Metric, + Page, PeriodicOptions, PeriodicTask, PublishOptions, @@ -117,6 +121,17 @@ export interface QueueOptions { * A Taskito queue: register tasks, enqueue work, read results, and run workers. * Backed by the Rust core over SQLite, Postgres, or Redis. */ +/** + * Normalize a native page to the declared {@link Page} shape. + * + * napi types an `Option` as `nextCursor?: string` but hands back `null` + * at runtime, so the two disagree: a `=== undefined` check against the raw value + * would silently never fire. Settle on `null`, matching the other SDKs. + */ +function toPage(page: { items: T[]; nextCursor?: string | null }): Page { + return { items: page.items, nextCursor: page.nextCursor ?? null }; +} + export class Queue { private readonly native: NativeQueue; private readonly serializer: Serializer; @@ -826,6 +841,57 @@ export class Queue { return this.native.listJobs(filter); } + /** + * List jobs on the wider filter: everything {@link Queue.listJobs} matches on, + * plus metadata/error substrings and a created-at range. + */ + listJobsFiltered(filter?: DetailedJobFilter): Promise { + return this.native.listJobsFiltered(filter); + } + + /** List archived (completed and moved out of the live table) jobs, newest first. */ + listArchived(limit?: number, offset?: number): Promise { + return this.native.listArchived(limit, offset); + } + + /** + * Keyset-paginated {@link Queue.listJobs}, ordered by created time. Pass a + * page's `nextCursor` back as `after`; `null` means the last page. + * + * O(page) at any depth on SQLite/Postgres. On Redis the status indexes are not + * seekable, so the keyset is applied in memory — correct, but O(matching rows). + */ + async listJobsAfter(filter?: CursorJobFilter, after?: string): Promise> { + return toPage(await this.native.listJobsAfter(filter, after)); + } + + /** + * Keyset-paginated {@link Queue.listJobsFiltered}, ordered by created time. + * See {@link Queue.listJobsAfter} for the cursor contract. + */ + async listJobsFilteredAfter( + filter?: CursorDetailedJobFilter, + after?: string, + ): Promise> { + return toPage(await this.native.listJobsFilteredAfter(filter, after)); + } + + /** + * Keyset-paginated {@link Queue.listArchived}, ordered by completed time. + * See {@link Queue.listJobsAfter} for the cursor contract. + */ + async listArchivedAfter(limit?: number, after?: string): Promise> { + return toPage(await this.native.listArchivedAfter(limit, after)); + } + + /** + * Keyset-paginated {@link Queue.deadLetters}, ordered by failed time. + * See {@link Queue.listJobsAfter} for the cursor contract. + */ + async deadLettersAfter(limit?: number, after?: string): Promise> { + return toPage(await this.native.deadLettersAfter(limit, after)); + } + /** Error history for a job (one entry per failed attempt). */ getJobErrors(id: string): Promise { return this.native.getJobErrors(id); diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts index 80db1fb6..3e7933fd 100644 --- a/sdks/node/src/types.ts +++ b/sdks/node/src/types.ts @@ -1,5 +1,7 @@ import type { CircuitBreakerInput, + DetailedJobFilter, + JobFilter, MeshWorkerConfig, EnqueueOptions as NativeEnqueueOptions, PublishOptions as NativePublishOptions, @@ -7,6 +9,7 @@ import type { export type { CircuitBreakerInput as CircuitBreakerOptions, + DetailedJobFilter, JobFilter, JsCircuitBreaker as CircuitBreaker, JsDagEdge as DagEdge, @@ -172,6 +175,28 @@ export interface RegisteredTask { options?: TaskOptions; } +/** + * One page of a keyset-paginated listing. + * + * `nextCursor` is `null` on the last page — a short page means the listing is + * exhausted — so a walk is `while (cursor !== null)`. Pass it back verbatim as + * the next call's `after`; treat it as opaque. + */ +export interface Page { + items: T[]; + nextCursor: string | null; +} + +/** + * A filter for a cursor-paginated listing: the offset-paged filter without its + * `offset`. The two ways of paging do not compose — a cursor already says where + * to resume — so the field is removed rather than silently ignored. + */ +export type CursorJobFilter = Omit; + +/** {@link DetailedJobFilter} without its `offset`. See {@link CursorJobFilter}. */ +export type CursorDetailedJobFilter = Omit; + /** Options for {@link Queue.runWorker}. */ export interface WorkerRunOptions { /** Queues to consume (default `["default"]`). */ diff --git a/sdks/node/test/core/listings.test.ts b/sdks/node/test/core/listings.test.ts new file mode 100644 index 00000000..cc257b43 --- /dev/null +++ b/sdks/node/test/core/listings.test.ts @@ -0,0 +1,66 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, it } from "vitest"; +import { Queue, type Worker } from "../../src/index"; + +let worker: Worker | undefined; + +afterEach(() => { + worker?.stop(); + worker = undefined; +}); + +function newQueue(): Queue { + return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-ls-")), "q.db") }); +} + +async function waitFor(predicate: () => Promise, timeoutMs = 14000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await predicate()) { + return true; + } + await new Promise((resolve) => setTimeout(resolve, 20)); + } + return false; +} + +it("filters jobs by metadata substring", async () => { + const queue = newQueue(); + queue.task("tagged", (n: number) => n); + + await queue.enqueue("tagged", [1], { metadata: "tenant=acme" }); + await queue.enqueue("tagged", [2], { metadata: "tenant=globex" }); + + const acme = await queue.listJobsFiltered({ metadataLike: "acme" }); + expect(acme).toHaveLength(1); + expect(acme[0]?.metadata).toBe("tenant=acme"); + + // The narrower listJobs has no metadata predicate, so it sees both. + expect(await queue.listJobs({ task: "tagged" })).toHaveLength(2); +}); + +it("rejects an unknown status on the wider filter too", async () => { + const queue = newQueue(); + await expect(queue.listJobsFiltered({ status: "nonsense" })).rejects.toThrow(/unknown status/); +}); + +it("lists a completed job from the archive", async () => { + // A terminal job is archived as it completes — the live `jobs` table only + // ever holds pending/running rows — so the archive is where a finished job + // is read back from, and purgeCompleted deletes from there rather than + // filling it. + const queue = newQueue(); + queue.task("done", (n: number) => n); + await queue.enqueue("done", [1]); + + worker = queue.runWorker({ queues: ["default"], concurrency: 2 }); + expect(await waitFor(async () => (await queue.stats()).completed >= 1)).toBe(true); + worker.stop(); + worker = undefined; + + const archived = await queue.listArchived(); + expect(archived).toHaveLength(1); + expect(archived[0]?.taskName).toBe("done"); +}); diff --git a/sdks/node/test/core/pagination.test.ts b/sdks/node/test/core/pagination.test.ts new file mode 100644 index 00000000..08075ebf --- /dev/null +++ b/sdks/node/test/core/pagination.test.ts @@ -0,0 +1,85 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { expect, it } from "vitest"; +import { type Job, type Page, Queue } from "../../src/index"; + +function newQueue(): Queue { + return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-pg-")), "q.db") }); +} + +/** Walk every page, collecting ids, so a dropped or repeated row shows up. */ +async function walk( + fetch: (after?: string) => Promise>, +): Promise<{ ids: string[]; pages: number }> { + const ids: string[] = []; + let cursor: string | undefined; + let pages = 0; + do { + const page: Page = await fetch(cursor); + ids.push(...page.items.map((job) => job.id)); + cursor = page.nextCursor ?? undefined; + pages += 1; + if (pages > 20) { + throw new Error("pagination did not terminate"); + } + } while (cursor !== undefined); + return { ids, pages }; +} + +it("walks every job exactly once across pages", async () => { + const queue = newQueue(); + queue.task("p", (n: number) => n); + const enqueued: string[] = []; + for (let i = 0; i < 5; i += 1) { + enqueued.push(queue.enqueue("p", [i])); + } + + const { ids, pages } = await walk((after) => queue.listJobsAfter({ limit: 2 }, after)); + + expect(pages).toBeGreaterThan(1); + expect(ids).toHaveLength(5); + expect(new Set(ids).size).toBe(5); + expect([...ids].sort()).toEqual([...enqueued].sort()); +}); + +it("ends the walk with a null cursor rather than looping", async () => { + const queue = newQueue(); + queue.task("p", (n: number) => n); + await queue.enqueue("p", [1]); + + // A page shorter than the limit is the last one. + const page = await queue.listJobsAfter({ limit: 10 }); + expect(page.items).toHaveLength(1); + expect(page.nextCursor).toBeNull(); +}); + +it("carries the wider filter across pages", async () => { + const queue = newQueue(); + queue.task("p", (n: number) => n); + for (let i = 0; i < 4; i += 1) { + await queue.enqueue("p", [i], { metadata: i < 3 ? "keep" : "drop" }); + } + + const { ids } = await walk((after) => + queue.listJobsFilteredAfter({ metadataLike: "keep", limit: 2 }, after), + ); + expect(ids).toHaveLength(3); +}); + +it("rejects an offset rather than silently returning a different page", async () => { + // Offset and cursor paging do not compose — a cursor already says where to + // resume — so honouring one and ignoring the other would quietly hand back a + // page the caller did not ask for. The TS type omits the field; this covers a + // caller who reaches past it. + const queue = newQueue(); + await expect(queue.listJobsAfter({ limit: 2, offset: 10 } as never)).rejects.toThrow( + /offset is not supported/, + ); +}); + +it("rejects a malformed cursor rather than restarting", async () => { + const queue = newQueue(); + await expect(queue.listJobsAfter({ limit: 2 }, "nocolon")).rejects.toThrow(/cursor/); + await expect(queue.deadLettersAfter(2, "-1:abc")).rejects.toThrow(/cursor/); +});