From 3c9a732497d4099d4c244801796998107f8e6599 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:05:43 +0530 Subject: [PATCH 1/8] feat(node): add listJobsFiltered and listArchived MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Python has had both; this SDK had neither, so a completed job could not be read back at all — terminal jobs are archived as they finish, so the archive is the only place they live. Both follow the existing scan-method shape (async + spawn_blocking). The wider filter takes its own input type, since it matches on metadata/error substrings and a created-at range that the narrow one has no fields for; the status parsing they share is now one helper rather than two copies. --- crates/taskito-node/src/config.rs | 22 +++++++ crates/taskito-node/src/queue/inspect.rs | 75 ++++++++++++++++++++---- sdks/node/src/index.ts | 1 + sdks/node/src/native.ts | 1 + sdks/node/src/queue.ts | 14 +++++ sdks/node/src/types.ts | 1 + sdks/node/test/core/listings.test.ts | 66 +++++++++++++++++++++ 7 files changed, 170 insertions(+), 10 deletions(-) create mode 100644 sdks/node/test/core/listings.test.ts 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/queue/inspect.rs b/crates/taskito-node/src/queue/inspect.rs index e1c96429..66fb23fb 100644 --- a/crates/taskito-node/src/queue/inspect.rs +++ b/crates/taskito-node/src/queue/inspect.rs @@ -9,7 +9,7 @@ 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, @@ -19,6 +19,17 @@ 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), + } +} + #[napi] impl JsQueue { /// Job counts by status across all queues. @@ -60,15 +71,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 +93,58 @@ 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)? + } + /// 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/sdks/node/src/index.ts b/sdks/node/src/index.ts index 6cfc2096..4e159745 100644 --- a/sdks/node/src/index.ts +++ b/sdks/node/src/index.ts @@ -80,6 +80,7 @@ export type { AnyHandler, CircuitBreakerOptions, DeadJob, + DetailedJobFilter, EnqueueOptions, Job, JobError, 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..efc4e119 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -51,6 +51,7 @@ import type { AnyHandler, CircuitBreaker, DeadJob, + DetailedJobFilter, EnqueueOptions, Job, JobDag, @@ -826,6 +827,19 @@ 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); + } + /** 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..daef2f83 100644 --- a/sdks/node/src/types.ts +++ b/sdks/node/src/types.ts @@ -7,6 +7,7 @@ import type { export type { CircuitBreakerInput as CircuitBreakerOptions, + DetailedJobFilter, JobFilter, JsCircuitBreaker as CircuitBreaker, JsDagEdge as DagEdge, 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"); +}); From 12b6d0cf297d2ae232ee3497123c2a7ea0bb6b0e Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:06:57 +0530 Subject: [PATCH 2/8] refactor(core): move next_cursor beside the cursor codec MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every SDK pages the same way, and "fewer rows than limit means last page" is part of the cursor contract rather than of any one shell — leaving it in the Python binding would have had the other two copy it. Pure move; Python now imports it. --- crates/taskito-core/src/storage/cursor.rs | 38 +++++++++++++++++++++++ crates/taskito-python/src/py_queue/mod.rs | 17 +--------- 2 files changed, 39 insertions(+), 16 deletions(-) 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-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] From d95574bae0e67fac1ff83cc20a03d0e8b60dada8 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:15:01 +0530 Subject: [PATCH 3/8] feat(node): add keyset pagination to the job and DLQ listings Python has had cursor paging since Tier 2; the core methods were already there. Adds listJobsAfter, listJobsFilteredAfter, listArchivedAfter and deadLettersAfter, so a deep listing stays O(page) instead of O(offset). napi has no generics, so each item type gets its own page object and the shell declares one Page they satisfy structurally. It also types an Option as optional but hands back null, so the shell normalizes: nextCursor is null on the last page, matching the other SDKs, and a `=== undefined` check against the raw value would never have fired. --- crates/taskito-node/src/convert/job.rs | 12 +++ crates/taskito-node/src/convert/mod.rs | 4 +- crates/taskito-node/src/convert/stats.rs | 8 ++ crates/taskito-node/src/queue/admin.rs | 34 ++++++- crates/taskito-node/src/queue/inspect.rs | 120 ++++++++++++++++++++++- sdks/node/src/index.ts | 1 + sdks/node/src/queue.ts | 47 +++++++++ sdks/node/src/types.ts | 12 +++ sdks/node/test/core/pagination.test.ts | 74 ++++++++++++++ 9 files changed, 308 insertions(+), 4 deletions(-) create mode 100644 sdks/node/test/core/pagination.test.ts 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 66fb23fb..01e05d13 100644 --- a/crates/taskito-node/src/queue/inspect.rs +++ b/crates/taskito-node/src/queue/inspect.rs @@ -13,7 +13,7 @@ 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}; @@ -30,6 +30,33 @@ fn parse_status_filter(status: Option<&str>) -> Result> { } } +/// 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. @@ -145,6 +172,97 @@ impl JsQueue { .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())?; + 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())?; + 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/sdks/node/src/index.ts b/sdks/node/src/index.ts index 4e159745..771fa8f5 100644 --- a/sdks/node/src/index.ts +++ b/sdks/node/src/index.ts @@ -87,6 +87,7 @@ export type { JobFilter, MeshWorkerConfig, Metric, + Page, PeriodicOptions, PeriodicTask, PublishOptions, diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index efc4e119..26791df7 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -58,6 +58,7 @@ import type { JobError, JobFilter, Metric, + Page, PeriodicOptions, PeriodicTask, PublishOptions, @@ -118,6 +119,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; @@ -840,6 +852,41 @@ export class Queue { 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?: JobFilter, 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?: DetailedJobFilter, 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 daef2f83..4158877d 100644 --- a/sdks/node/src/types.ts +++ b/sdks/node/src/types.ts @@ -173,6 +173,18 @@ 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; +} + /** Options for {@link Queue.runWorker}. */ export interface WorkerRunOptions { /** Queues to consume (default `["default"]`). */ diff --git a/sdks/node/test/core/pagination.test.ts b/sdks/node/test/core/pagination.test.ts new file mode 100644 index 00000000..9ed7be23 --- /dev/null +++ b/sdks/node/test/core/pagination.test.ts @@ -0,0 +1,74 @@ +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 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/); +}); From 84bd6ba2df50fcb161960eaf8cc08215c518976d Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:27:12 +0530 Subject: [PATCH 4/8] feat(java): add keyset pagination to the job listings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Python has had cursor paging since Tier 2 and the core methods were already there. Adds listJobsAfter and listArchivedAfter with a Page, so a deep listing stays O(page) rather than O(offset). The SPI methods are default: QueueBackend has an implementor outside this module, and an abstract method would break it and every downstream backend. They throw rather than return an empty page — a backend that cannot seek would otherwise look like it had reached the last page. The in-memory test backend implements them for real, since a default-only method is untested API. The smoke exercises a paginated call: GraalVM metadata is generated from what runs, so a DTO it never touches gets no reflection entry and fails under --no-fallback at runtime rather than in CI. --- crates/taskito-java/src/convert.rs | 9 ++ crates/taskito-java/src/queue/inspect.rs | 82 +++++++++++++++- .../org/byteveda/taskito/graalvm/Smoke.java | 8 +- .../org/byteveda/taskito/DefaultTaskito.java | 20 ++++ .../java/org/byteveda/taskito/Taskito.java | 24 +++++ .../taskito/internal/JniQueueBackend.java | 10 ++ .../taskito/internal/NativeQueue.java | 4 + .../java/org/byteveda/taskito/model/Page.java | 42 ++++++++ .../byteveda/taskito/spi/QueueBackend.java | 14 +++ .../byteveda/taskito/core/PaginationTest.java | 96 +++++++++++++++++++ .../taskito/test/InMemoryQueueBackend.java | 78 +++++++++++++++ 11 files changed, 385 insertions(+), 2 deletions(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/model/Page.java create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/core/PaginationTest.java 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/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..17c27909 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -36,6 +36,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 +469,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 +1066,15 @@ private static R decode(String json, Class type) { } } + 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..a123765b 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. */ 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/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..9254d49d --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/Page.java @@ -0,0 +1,42 @@ +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
+ * String cursor = null;
+ * do {
+ *     Page page = queue.listJobsAfter(filter, 50, cursor);
+ *     page.items.forEach(::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/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/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..9c3b388e 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,84 @@ 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 = 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); + } + } + return keysetPage( + archived, + limit > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) limit, + 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 "[]"; From bb90f88de0d60d271cd2d9bb7c336164df277a00 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:37:01 +0530 Subject: [PATCH 5/8] feat(java): add per-task middleware disables Python and Node can turn a middleware off for one task without a redeploy; this SDK had only the global use(). Adds disableMiddleware/enableMiddleware/ listDisabledMiddleware over the existing settings, keyed the same way as the other SDKs. The list is read per invocation rather than cached, like Node, so a toggle takes effect on the next job with no restart. The chain is resolved once and reused for before/after/onError: re-reading between them would let a toggle landing mid-job run after on a middleware whose before never ran. The key format and the name derivation live in one place, so the worker that honours the list and the API that writes it cannot drift apart. --- .../org/byteveda/taskito/DefaultTaskito.java | 23 ++++ .../java/org/byteveda/taskito/Taskito.java | 29 ++++ .../taskito/internal/MiddlewareDisables.java | 84 ++++++++++++ .../taskito/worker/WorkerDispatchBridge.java | 15 +- .../taskito/worker/MiddlewareDisableTest.java | 129 ++++++++++++++++++ 5 files changed, 276 insertions(+), 4 deletions(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/internal/MiddlewareDisables.java create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/worker/MiddlewareDisableTest.java 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 17c27909..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; @@ -1066,6 +1067,28 @@ 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 { 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 a123765b..803ff455 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java @@ -255,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/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/worker/WorkerDispatchBridge.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java index b64528de..c6a0ccd1 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,22 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { if (scope != null) { Resources.enter(scope); } + // 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. + List chain = disables.resolve(taskName, middleware); try { - for (Middleware m : 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 +139,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/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"); + } +} From 17de5420e02ef4c90f0cf19eae1405f33e4677a4 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:55:51 +0530 Subject: [PATCH 6/8] fix(java): resolve the middleware chain inside the job's try block MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resolving reads the backend, so it can throw — and it ran after the resource scope was bound but before the try, leaving the scope entered, the finally unreached and the job neither completed nor failed until the reaper timed it out. A settings failure now fails the job like any other. The chain starts empty so that failure runs onError on nothing, which is right: no before() ran either. --- .../taskito/worker/WorkerDispatchBridge.java | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) 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 c6a0ccd1..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 @@ -93,11 +93,16 @@ private void runJob(long token, String jobId, String taskName, byte[] payload) { if (scope != null) { Resources.enter(scope); } - // 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. - List chain = disables.resolve(taskName, middleware); + // 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 { + // 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); } From fbfb9bfb61233224f8461f0472d4716c29cc7081 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:55:51 +0530 Subject: [PATCH 7/8] fix(node): reject an offset on a cursor-paginated call MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The keyset methods took the offset-paged filter and ignored its offset, so a caller passing one got a different page than they asked for, silently. Offset and cursor paging do not compose — a cursor already says where to resume. The TS filters omit the field, and the native call rejects a non-zero one for a caller who reaches past the type. --- crates/taskito-node/src/queue/inspect.rs | 14 ++++++++++++++ sdks/node/src/index.ts | 2 ++ sdks/node/src/queue.ts | 9 +++++++-- sdks/node/src/types.ts | 12 ++++++++++++ sdks/node/test/core/pagination.test.ts | 11 +++++++++++ 5 files changed, 46 insertions(+), 2 deletions(-) diff --git a/crates/taskito-node/src/queue/inspect.rs b/crates/taskito-node/src/queue/inspect.rs index 01e05d13..c571bbc4 100644 --- a/crates/taskito-node/src/queue/inspect.rs +++ b/crates/taskito-node/src/queue/inspect.rs @@ -30,6 +30,18 @@ fn parse_status_filter(status: Option<&str>) -> Result> { } } +/// 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> { @@ -186,6 +198,7 @@ impl JsQueue { ) -> 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(); @@ -217,6 +230,7 @@ impl JsQueue { ) -> 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(); diff --git a/sdks/node/src/index.ts b/sdks/node/src/index.ts index 771fa8f5..a5a59b8e 100644 --- a/sdks/node/src/index.ts +++ b/sdks/node/src/index.ts @@ -79,6 +79,8 @@ export { decodeTaskError, encodeTaskError, type TaskError } from "./task-error"; export type { AnyHandler, CircuitBreakerOptions, + CursorDetailedJobFilter, + CursorJobFilter, DeadJob, DetailedJobFilter, EnqueueOptions, diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index 26791df7..a59abc3d 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -50,6 +50,8 @@ import { import type { AnyHandler, CircuitBreaker, + CursorDetailedJobFilter, + CursorJobFilter, DeadJob, DetailedJobFilter, EnqueueOptions, @@ -859,7 +861,7 @@ export class Queue { * 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?: JobFilter, after?: string): Promise> { + async listJobsAfter(filter?: CursorJobFilter, after?: string): Promise> { return toPage(await this.native.listJobsAfter(filter, after)); } @@ -867,7 +869,10 @@ export class Queue { * Keyset-paginated {@link Queue.listJobsFiltered}, ordered by created time. * See {@link Queue.listJobsAfter} for the cursor contract. */ - async listJobsFilteredAfter(filter?: DetailedJobFilter, after?: string): Promise> { + async listJobsFilteredAfter( + filter?: CursorDetailedJobFilter, + after?: string, + ): Promise> { return toPage(await this.native.listJobsFilteredAfter(filter, after)); } diff --git a/sdks/node/src/types.ts b/sdks/node/src/types.ts index 4158877d..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, @@ -185,6 +187,16 @@ export interface Page { 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/pagination.test.ts b/sdks/node/test/core/pagination.test.ts index 9ed7be23..08075ebf 100644 --- a/sdks/node/test/core/pagination.test.ts +++ b/sdks/node/test/core/pagination.test.ts @@ -67,6 +67,17 @@ it("carries the wider filter across pages", async () => { 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/); From ccc2118154490de17bb68f8f83d2b5264e211809 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:55:58 +0530 Subject: [PATCH 8/8] fix(java): clamp a negative page limit, correct the Page example subList would throw on a negative limit where the native handlers clamp it to an empty page, so the in-memory backend diverged from the contract it exists to stand in for. The Page javadoc called listJobsAfter with a limit argument it does not take. --- .../src/main/java/org/byteveda/taskito/model/Page.java | 5 +++-- .../org/byteveda/taskito/test/InMemoryQueueBackend.java | 7 +++++-- 2 files changed, 8 insertions(+), 4 deletions(-) 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 index 9254d49d..309d8f90 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/model/Page.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/Page.java @@ -12,10 +12,11 @@ * 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, 50, cursor);
- *     page.items.forEach(::handle);
+ *     Page page = queue.listJobsAfter(filter, cursor);
+ *     page.items.forEach(this::handle);
  *     cursor = page.nextCursor;
  * } while (cursor != null);
  * }
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 9c3b388e..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 @@ -221,7 +221,7 @@ public String listJobsAfterJson(String filterJson, String afterOrNull) { String status = text(filter, "status"); String queue = text(filter, "queue"); String task = text(filter, "task"); - int limit = optInt(filter, "limit", Integer.MAX_VALUE); + int limit = Math.max(0, optInt(filter, "limit", Integer.MAX_VALUE)); List matching = new ArrayList<>(); for (JobRec job : jobs.values()) { @@ -247,9 +247,12 @@ public String listArchivedAfterJson(long limit, String afterOrNull) { 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, - limit > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) limit, + (int) Math.min(clamped, Integer.MAX_VALUE), afterOrNull, job -> job.completedAt == null ? job.createdAt : job.completedAt); }