perf: pagination and bounded scans at millions of rows

crates/taskito-core/Cargo.toml

@@ -37,3 +37,8 @@ tempfile = "3"
 # state and assert on internal index keys. Only used by `#[cfg(feature = "redis")]`
 # tests; harmless when the feature is off.
 redis = { workspace = true }
+# Raw connection access in the Postgres contract test to reset the `taskito`
+# schema up front, so the count-exact assertions are deterministic on a
+# persistent test DB (the Postgres analogue of the Redis suite's FLUSHDB). Only
+# used by `#[cfg(feature = "postgres")]` tests.
+diesel = { workspace = true }

crates/taskito-core/src/job.rs

@@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use uuid::Uuid;
 
-use crate::storage::models::{ArchivedJobRow, JobRow, NarrowJobRow};
+use crate::storage::models::{ArchivedJobRow, JobRow, NarrowArchivedJobRow, NarrowJobRow};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[repr(i32)]
@@ -190,6 +190,39 @@ impl Job {
             has_deps: narrow.has_deps,
         }
     }
+
+    /// Assemble a terminal [`Job`] from a blob-free [`NarrowArchivedJobRow`].
+    /// Listing paths use this so paging the archive never loads `payload`/
+    /// `result`; both come back empty (fetch the full job via `get_job`).
+    /// Archived jobs are terminal and never re-dequeued, so `has_deps` is false.
+    pub fn from_narrow_archived(narrow: NarrowArchivedJobRow) -> Self {
+        Self {
+            id: narrow.id,
+            queue: narrow.queue,
+            task_name: narrow.task_name,
+            payload: Vec::new(),
+            status: JobStatus::from_i32(narrow.status).unwrap_or(JobStatus::Pending),
+            priority: narrow.priority,
+            created_at: narrow.created_at,
+            scheduled_at: narrow.scheduled_at,
+            started_at: narrow.started_at,
+            completed_at: narrow.completed_at,
+            retry_count: narrow.retry_count,
+            max_retries: narrow.max_retries,
+            result: None,
+            error: narrow.error,
+            timeout_ms: narrow.timeout_ms,
+            unique_key: narrow.unique_key,
+            progress: narrow.progress,
+            metadata: narrow.metadata,
+            notes: narrow.notes,
+            cancel_requested: narrow.cancel_requested != 0,
+            expires_at: narrow.expires_at,
+            result_ttl_ms: narrow.result_ttl_ms,
+            namespace: narrow.namespace,
+            has_deps: false,
+        }
+    }
 }
 
 /// A successful job outcome to persist. Batches the three writes the success

crates/taskito-core/src/storage/cursor.rs

@@ -0,0 +1,53 @@
+//! Opaque keyset-pagination cursor shared by every SDK binding.
+//!
+//! A cursor encodes the `(sort_key, id)` of the last row of a page — the sort
+//! key being the listing's ordering column (`created_at`/`completed_at`/
+//! `failed_at`, a non-negative millis timestamp) and `id` the row's UUIDv7 /
+//! DLQ id. Neither contains a `:`, so a single split round-trips unambiguously
+//! with no base64. Callers must treat the string as opaque and pass it back
+//! verbatim; the format may gain a version prefix later.
+
+use crate::error::{QueueError, Result};
+
+/// Encode a `(sort_key, id)` keyset cursor as an opaque string.
+pub fn encode_cursor(sort_key: i64, id: &str) -> String {
+    format!("{sort_key}:{id}")
+}
+
+/// Decode a cursor produced by [`encode_cursor`]. Returns `Other` on malformed
+/// input — treat it like any other bad-request error.
+pub fn decode_cursor(cursor: &str) -> Result<(i64, &str)> {
+    let (key, id) = cursor
+        .split_once(':')
+        .ok_or_else(|| QueueError::Other(format!("invalid cursor: {cursor}")))?;
+    let sort_key: i64 = key
+        .parse()
+        .map_err(|_| QueueError::Other(format!("invalid cursor: {cursor}")))?;
+    // A cursor this crate did not encode: sort keys are non-negative millis and
+    // every row has an id. Reject rather than page from a position no row holds.
+    if sort_key < 0 || id.is_empty() {
+        return Err(QueueError::Other(format!("invalid cursor: {cursor}")));
+    }
+    Ok((sort_key, id))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn round_trips() {
+        let s = encode_cursor(1_720_000_000_123, "0190a1b2-c3d4-7e5f-8a9b-0c1d2e3f4a5b");
+        let (key, id) = decode_cursor(&s).unwrap();
+        assert_eq!(key, 1_720_000_000_123);
+        assert_eq!(id, "0190a1b2-c3d4-7e5f-8a9b-0c1d2e3f4a5b");
+    }
+
+    #[test]
+    fn rejects_malformed() {
+        assert!(decode_cursor("nocolon").is_err());
+        assert!(decode_cursor("notanint:abc").is_err());
+        assert!(decode_cursor("-1:abc").is_err());
+        assert!(decode_cursor("1:").is_err());
+    }
+}

crates/taskito-core/src/storage/diesel_common/archival.rs

@@ -9,14 +9,53 @@ macro_rules! impl_diesel_archival_ops {
             pub fn list_archived(&self, limit: i64, offset: i64) -> Result<Vec<Job>> {
                 let mut conn = self.conn()?;
 
-                let rows: Vec<ArchivedJobRow> = archived_jobs::table
+                // Narrow projection: archive listings never render the
+                // arg/result blobs, so leave them on TOAST/overflow pages.
+                let rows: Vec<NarrowArchivedJobRow> = archived_jobs::table
                     .order(archived_jobs::completed_at.desc())
                     .limit(limit)
                     .offset(offset)
-                    .select(ArchivedJobRow::as_select())
+                    .select(NarrowArchivedJobRow::as_select())
                     .load(&mut conn)?;
 
-                Ok(rows.into_iter().map(Job::from).collect())
+                Ok(rows.into_iter().map(Job::from_narrow_archived).collect())
+            }
+
+            /// Keyset-paginated `list_archived`, ordered by
+            /// `(completed_at, id)` descending. `completed_at` is nullable in
+            /// the schema (always set for a terminal row), so the cursor bound
+            /// compares against `Some(..)`.
+            pub fn list_archived_after(
+                &self,
+                limit: i64,
+                after: Option<(i64, &str)>,
+            ) -> Result<Vec<Job>> {
+                if limit <= 0 {
+                    return Ok(Vec::new());
+                }
+                let mut conn = self.conn()?;
+
+                let mut query = archived_jobs::table
+                    .into_boxed()
+                    .order((archived_jobs::completed_at.desc(), archived_jobs::id.desc()));
+
+                if let Some((cursor_completed_at, cursor_id)) = after {
+                    let cursor_id = cursor_id.to_string();
+                    query = query.filter(
+                        archived_jobs::completed_at
+                            .lt(Some(cursor_completed_at))
+                            .or(archived_jobs::completed_at
+                                .eq(Some(cursor_completed_at))
+                                .and(archived_jobs::id.lt(cursor_id))),
+                    );
+                }
+
+                let rows: Vec<NarrowArchivedJobRow> = query
+                    .limit(limit)
+                    .select(NarrowArchivedJobRow::as_select())
+                    .load(&mut conn)?;
+
+                Ok(rows.into_iter().map(Job::from_narrow_archived).collect())
             }
         }
     };

crates/taskito-core/src/storage/diesel_common/dead_letter.rs

@@ -90,14 +90,50 @@ macro_rules! impl_diesel_dead_letter_ops {
             pub fn list_dead(&self, limit: i64, offset: i64) -> Result<Vec<DeadJob>> {
                 let mut conn = self.conn()?;
 
-                let rows: Vec<DeadLetterRow> = dead_letter::table
+                // Narrow projection: DLQ listings never render the arg blob.
+                let rows: Vec<NarrowDeadLetterRow> = dead_letter::table
                     .order(dead_letter::failed_at.desc())
                     .limit(limit)
                     .offset(offset)
-                    .select(DeadLetterRow::as_select())
+                    .select(NarrowDeadLetterRow::as_select())
+                    .load(&mut conn)?;
+
+                Ok(rows.into_iter().map(DeadJob::from_narrow).collect())
+            }
+
+            /// Keyset-paginated `list_dead`, ordered by `(failed_at, id)`
+            /// descending with a `(failed_at, id) < cursor` bound.
+            pub fn list_dead_after(
+                &self,
+                limit: i64,
+                after: Option<(i64, &str)>,
+            ) -> Result<Vec<DeadJob>> {
+                if limit <= 0 {
+                    return Ok(Vec::new());
+                }
+                let mut conn = self.conn()?;
+
+                let mut query = dead_letter::table
+                    .into_boxed()
+                    .order((dead_letter::failed_at.desc(), dead_letter::id.desc()));
+
+                if let Some((cursor_failed_at, cursor_id)) = after {
+                    let cursor_id = cursor_id.to_string();
+                    query = query.filter(
+                        dead_letter::failed_at
+                            .lt(cursor_failed_at)
+                            .or(dead_letter::failed_at
+                                .eq(cursor_failed_at)
+                                .and(dead_letter::id.lt(cursor_id))),
+                    );
+                }
+
+                let rows: Vec<NarrowDeadLetterRow> = query
+                    .limit(limit)
+                    .select(NarrowDeadLetterRow::as_select())
                     .load(&mut conn)?;
 
-                Ok(rows.into_iter().map(DeadJob::from).collect())
+                Ok(rows.into_iter().map(DeadJob::from_narrow).collect())
             }
 
             /// List dead letter entries for a single task, newest first.
@@ -117,15 +153,15 @@ macro_rules! impl_diesel_dead_letter_ops {
 
                 let mut conn = self.conn()?;
 
-                let rows: Vec<DeadLetterRow> = dead_letter::table
+                let rows: Vec<NarrowDeadLetterRow> = dead_letter::table
                     .filter(dead_letter::task_name.eq(task_name))
                     .order(dead_letter::failed_at.desc())
                     .limit(limit)
                     .offset(offset)
-                    .select(DeadLetterRow::as_select())
+                    .select(NarrowDeadLetterRow::as_select())
                     .load(&mut conn)?;
 
-                Ok(rows.into_iter().map(DeadJob::from).collect())
+                Ok(rows.into_iter().map(DeadJob::from_narrow).collect())
             }
 
             /// Delete every dead letter entry for a task. Returns the count removed.
@@ -321,13 +357,16 @@ macro_rules! impl_diesel_dead_letter_ops {
                     None => query = query.filter(dead_letter::namespace.is_null()),
                 }
 
-                let rows: Vec<DeadLetterRow> = query
+                // Narrow projection: the auto-retry loop reads only `id`/
+                // `dlq_retry_count`; `retry_dead(id)` re-reads the full row to
+                // rebuild the payload, so listing never needs the arg blob.
+                let rows: Vec<NarrowDeadLetterRow> = query
                     .order(dead_letter::failed_at.asc())
                     .limit(limit)
-                    .select(DeadLetterRow::as_select())
+                    .select(NarrowDeadLetterRow::as_select())
                     .load(&mut conn)?;
 
-                Ok(rows.into_iter().map(DeadJob::from).collect())
+                Ok(rows.into_iter().map(DeadJob::from_narrow).collect())
             }
         }
     };