From 12346d422c8882ed77149bd78478ceac64c53149 Mon Sep 17 00:00:00 2001 From: geofmureithi Date: Mon, 20 Oct 2025 18:42:24 +0300 Subject: [PATCH 1/3] bump: to 1.0.0-alpha.2 --- Cargo.toml | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 80e7c07..5835b35 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,11 +26,11 @@ features = ["chrono", "sqlite"] [dependencies] serde = { version = "1", features = ["derive"] } serde_json = { version = "1" } -apalis-core = { version = "1.0.0-alpha.4", default-features = false, features = [ +apalis-core = { version = "1.0.0-alpha.6", default-features = false, features = [ "sleep", "json", -], git = "https://github.com/geofmureithi/apalis.git", branch = "chore/traits-expansion" } -apalis-workflow = { version = "0.1.0-alpha.3", git = "https://github.com/geofmureithi/apalis.git", branch = "chore/traits-expansion" } +] } +apalis-workflow = { version = "0.1.0-alpha.4" } log = "0.4.21" futures = "0.3.30" tokio = { version = "1", features = ["rt", "net"], optional = true } @@ -46,10 +46,8 @@ bytes = "1.1.0" [dev-dependencies] tokio = { version = "1", features = ["macros", "rt-multi-thread"] } -apalis-core = { version = "1.0.0-alpha.4", features = [ - "test-utils", -], git = "https://github.com/geofmureithi/apalis.git", branch = "chore/traits-expansion" } -apalis-workflow = { version = "0.1.0-alpha.2", git = "https://github.com/geofmureithi/apalis.git", branch = "chore/traits-expansion" } +apalis-core = { version = "1.0.0-alpha.6", features = ["test-utils"] } +apalis-workflow = { version = "0.1.0-alpha.4" } apalis-sqlite = { path = ".", features = ["migrate", "tokio-comp"] } [package.metadata.docs.rs] From e600e4500b9e26c83dbe1a74023678d92fa3401a Mon Sep 17 00:00:00 2001 From: geofmureithi Date: Fri, 24 Oct 2025 11:13:47 +0300 Subject: [PATCH 2/3] chore: move to bytes --- ...1c94a3743d2a7608360446ecd82df3264f4ce.json | 12 ++ Cargo.toml | 17 +- README.md | 4 +- .../20220530084123_jobs_workers.sql | 0 .../20250313213411_add_job_priority.sql | 0 .../20251013233016_stats_indexes.sql | 0 .../20251017150712_rename_last_error.sql | 0 .../20251017162501_worker_started_at.sql | 0 .../{json => }/20251018162501_metadata.sql | 0 ...e.sql => 20251018164941_move_to_bytes.sql} | 57 +++++-- .../json/20251017162501_worker_started_at.sql | 4 - src/ack.rs | 11 +- src/config.rs | 139 +--------------- src/context.rs | 152 ------------------ src/fetcher.rs | 13 +- src/from_row.rs | 7 +- src/lib.rs | 111 +++++++++---- src/queries/fetch_by_id.rs | 17 +- src/queries/list_queues.rs | 4 +- src/queries/list_tasks.rs | 23 ++- src/queries/list_workers.rs | 4 +- src/queries/metrics.rs | 4 +- src/queries/mod.rs | 1 + src/queries/vacuum.rs | 20 +++ src/queries/wait_for.rs | 2 +- src/shared.rs | 14 +- 26 files changed, 236 insertions(+), 380 deletions(-) create mode 100644 .sqlx/query-b3e8f3efeeb73cc19602247f4d61c94a3743d2a7608360446ecd82df3264f4ce.json rename migrations/{json => }/20220530084123_jobs_workers.sql (100%) rename migrations/{json => }/20250313213411_add_job_priority.sql (100%) rename migrations/{json => }/20251013233016_stats_indexes.sql (100%) rename migrations/{json => }/20251017150712_rename_last_error.sql (100%) rename migrations/{bytes => }/20251017162501_worker_started_at.sql (100%) rename migrations/{json => }/20251018162501_metadata.sql (100%) rename migrations/{bytes/20251017141210_initial_bytes_storage.sql => 20251018164941_move_to_bytes.sql} (70%) delete mode 100644 migrations/json/20251017162501_worker_started_at.sql delete mode 100644 src/context.rs create mode 100644 src/queries/vacuum.rs diff --git a/.sqlx/query-b3e8f3efeeb73cc19602247f4d61c94a3743d2a7608360446ecd82df3264f4ce.json b/.sqlx/query-b3e8f3efeeb73cc19602247f4d61c94a3743d2a7608360446ecd82df3264f4ce.json new file mode 100644 index 0000000..3f5c458 --- /dev/null +++ b/.sqlx/query-b3e8f3efeeb73cc19602247f4d61c94a3743d2a7608360446ecd82df3264f4ce.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "Delete from\n Jobs\nwhere\n status = 'Done'\n OR status = 'Killed'\n OR (\n status = 'Failed'\n AND max_attempts <= attempts\n );\n\nVACUUM;\n", + "describe": { + "columns": [], + "parameters": { + "Right": 0 + }, + "nullable": [] + }, + "hash": "b3e8f3efeeb73cc19602247f4d61c94a3743d2a7608360446ecd82df3264f4ce" +} diff --git a/Cargo.toml b/Cargo.toml index 5835b35..2522107 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,8 +15,7 @@ async-std-comp = ["async-std", "sqlx/runtime-async-std-rustls"] async-std-comp-native-tls = ["async-std", "sqlx/runtime-async-std-native-tls"] tokio-comp = ["tokio", "sqlx/runtime-tokio-rustls"] tokio-comp-native-tls = ["tokio", "sqlx/runtime-tokio-native-tls"] -bytes = [] -json = ["sqlx/json"] +json = ["apalis-core/json", "sqlx/json"] [dependencies.sqlx] version = "0.8.6" @@ -28,9 +27,9 @@ serde = { version = "1", features = ["derive"] } serde_json = { version = "1" } apalis-core = { version = "1.0.0-alpha.6", default-features = false, features = [ "sleep", - "json", -] } -apalis-workflow = { version = "0.1.0-alpha.4" } +], path = "../apalis/packages/apalis-core" } +apalis-sql = { version = "1.0.0-alpha.6", path = "../apalis/packages/apalis-sql" } +apalis-workflow = { version = "0.1.0-alpha.4", path = "../apalis/packages/apalis-workflow" } log = "0.4.21" futures = "0.3.30" tokio = { version = "1", features = ["rt", "net"], optional = true } @@ -46,11 +45,13 @@ bytes = "1.1.0" [dev-dependencies] tokio = { version = "1", features = ["macros", "rt-multi-thread"] } -apalis-core = { version = "1.0.0-alpha.6", features = ["test-utils"] } -apalis-workflow = { version = "0.1.0-alpha.4" } +apalis-core = { version = "1.0.0-alpha.6", features = [ + "test-utils", +], path = "../apalis/packages/apalis-core" } +apalis-workflow = { version = "0.1.0-alpha.4", path = "../apalis/packages/apalis-workflow" } apalis-sqlite = { path = ".", features = ["migrate", "tokio-comp"] } [package.metadata.docs.rs] # defines the configuration attribute `docsrs` rustdoc-args = ["--cfg", "docsrs"] -features = ["migrate", "tokio-comp", "json"] +all-features = true diff --git a/README.md b/README.md index ff2c9ef..ca9a3c3 100644 --- a/README.md +++ b/README.md @@ -34,7 +34,7 @@ async fn main() { start += 1; let task = Task::builder(start) .run_after(Duration::from_secs(1)) - .with_ctx(SqliteContext::new().with_priority(1)) + .with_ctx(SqlContext::new().with_priority(1)) .build(); Ok(task) }) @@ -79,7 +79,7 @@ async fn main() { start += 1; Task::builder(serde_json::to_value(&start).unwrap()) .run_after(Duration::from_secs(1)) - .with_ctx(SqliteContext::new().with_priority(start)) + .with_ctx(SqlContext::new().with_priority(start)) .build() }) .take(20) diff --git a/migrations/json/20220530084123_jobs_workers.sql b/migrations/20220530084123_jobs_workers.sql similarity index 100% rename from migrations/json/20220530084123_jobs_workers.sql rename to migrations/20220530084123_jobs_workers.sql diff --git a/migrations/json/20250313213411_add_job_priority.sql b/migrations/20250313213411_add_job_priority.sql similarity index 100% rename from migrations/json/20250313213411_add_job_priority.sql rename to migrations/20250313213411_add_job_priority.sql diff --git a/migrations/json/20251013233016_stats_indexes.sql b/migrations/20251013233016_stats_indexes.sql similarity index 100% rename from migrations/json/20251013233016_stats_indexes.sql rename to migrations/20251013233016_stats_indexes.sql diff --git a/migrations/json/20251017150712_rename_last_error.sql b/migrations/20251017150712_rename_last_error.sql similarity index 100% rename from migrations/json/20251017150712_rename_last_error.sql rename to migrations/20251017150712_rename_last_error.sql diff --git a/migrations/bytes/20251017162501_worker_started_at.sql b/migrations/20251017162501_worker_started_at.sql similarity index 100% rename from migrations/bytes/20251017162501_worker_started_at.sql rename to migrations/20251017162501_worker_started_at.sql diff --git a/migrations/json/20251018162501_metadata.sql b/migrations/20251018162501_metadata.sql similarity index 100% rename from migrations/json/20251018162501_metadata.sql rename to migrations/20251018162501_metadata.sql diff --git a/migrations/bytes/20251017141210_initial_bytes_storage.sql b/migrations/20251018164941_move_to_bytes.sql similarity index 70% rename from migrations/bytes/20251017141210_initial_bytes_storage.sql rename to migrations/20251018164941_move_to_bytes.sql index 97836d8..de2547b 100644 --- a/migrations/bytes/20251017141210_initial_bytes_storage.sql +++ b/migrations/20251018164941_move_to_bytes.sql @@ -1,16 +1,5 @@ -CREATE TABLE IF NOT EXISTS Workers ( - id TEXT NOT NULL UNIQUE, - worker_type TEXT NOT NULL, - storage_name TEXT NOT NULL, - layers TEXT, - last_seen INTEGER NOT NULL DEFAULT (strftime('%s', 'now')) -); - -CREATE INDEX IF NOT EXISTS Idx ON Workers(id); - -CREATE INDEX IF NOT EXISTS WTIdx ON Workers(worker_type); - -CREATE INDEX IF NOT EXISTS LSIdx ON Workers(last_seen); +ALTER TABLE + Jobs RENAME TO Jobs_old; CREATE TABLE IF NOT EXISTS Jobs ( job BLOB NOT NULL, @@ -24,9 +13,47 @@ CREATE TABLE IF NOT EXISTS Jobs ( lock_at INTEGER, lock_by TEXT, done_at INTEGER, + priority INTEGER NOT NULL DEFAULT 0, + metadata TEXT, + PRIMARY KEY(id), FOREIGN KEY(lock_by) REFERENCES Workers(id) ); +INSERT INTO + Jobs ( + job, + id, + job_type, + status, + attempts, + max_attempts, + run_at, + last_result, + lock_at, + lock_by, + done_at, + priority, + metadata + ) +SELECT + CAST(job AS BLOB), + id, + job_type, + status, + attempts, + max_attempts, + run_at, + last_result, + lock_at, + lock_by, + done_at, + priority, + metadata +FROM + Jobs_old; + +DROP TABLE Jobs_old; + CREATE INDEX IF NOT EXISTS TIdx ON Jobs(id); CREATE INDEX IF NOT EXISTS SIdx ON Jobs(status); @@ -35,10 +62,6 @@ CREATE INDEX IF NOT EXISTS LIdx ON Jobs(lock_by); CREATE INDEX IF NOT EXISTS JTIdx ON Jobs(job_type); - -ALTER TABLE Jobs -ADD priority INTEGER NOT NULL DEFAULT 0; - CREATE INDEX IF NOT EXISTS idx_jobs_job_type_status_run_at ON Jobs(job_type, status, run_at); CREATE INDEX IF NOT EXISTS idx_jobs_status_run_at ON Jobs(status, run_at); diff --git a/migrations/json/20251017162501_worker_started_at.sql b/migrations/json/20251017162501_worker_started_at.sql deleted file mode 100644 index 1ef26e5..0000000 --- a/migrations/json/20251017162501_worker_started_at.sql +++ /dev/null @@ -1,4 +0,0 @@ -ALTER TABLE - Workers -ADD - COLUMN started_at INTEGER; diff --git a/src/ack.rs b/src/ack.rs index 6a27027..e96c1c5 100644 --- a/src/ack.rs +++ b/src/ack.rs @@ -5,6 +5,7 @@ use apalis_core::{ task::{Parts, status::Status}, worker::{context::WorkerContext, ext::ack::Acknowledge}, }; +use apalis_sql::context::SqlContext; use apalis_workflow::StepResult; use futures::{FutureExt, future::BoxFuture}; use serde::Serialize; @@ -14,7 +15,7 @@ use tower_layer::Layer; use tower_service::Service; use ulid::Ulid; -use crate::{CompactType, SqliteTask, context::SqliteContext}; +use crate::{CompactType, SqliteTask}; #[derive(Clone)] pub struct SqliteAck { @@ -26,13 +27,13 @@ impl SqliteAck { } } -impl Acknowledge for SqliteAck { +impl Acknowledge for SqliteAck { type Error = sqlx::Error; type Future = BoxFuture<'static, Result<(), Self::Error>>; fn ack( &mut self, res: &Result, - parts: &Parts, + parts: &Parts, ) -> Self::Future { let task_id = parts.task_id; let worker_id = parts.ctx.lock_by().clone(); @@ -42,7 +43,7 @@ impl Acknowledge for SqliteA Ok(r) => { if let Some(res_ref) = (r as &dyn Any).downcast_ref::>() { let res_deserialized: Result = - serde_json::from_str(&res_ref.0); + serde_json::from_slice(&res_ref.0); serde_json::to_string(&res_deserialized.map_err(|e| e.to_string())) } else { serde_json::to_string(&res.as_ref().map_err(|e| e.to_string())) @@ -85,7 +86,7 @@ impl Acknowledge for SqliteA } pub fn calculate_status( - parts: &Parts, + parts: &Parts, res: &Result, ) -> Status { match &res { diff --git a/src/config.rs b/src/config.rs index fa8aeb2..b65f474 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,145 +1,16 @@ -use std::time::Duration; - -use apalis_core::backend::{ - Backend, ConfigExt, - poll_strategy::{BackoffConfig, IntervalStrategy, MultiStrategy, StrategyBuilder}, - queue::Queue, -}; +use apalis_core::backend::{Backend, ConfigExt, queue::Queue}; use ulid::Ulid; -use crate::{CompactType, SqliteContext, SqliteStorage}; - -#[derive(Debug, Clone)] -pub struct Config { - keep_alive: Duration, - buffer_size: usize, - poll_strategy: MultiStrategy, - reenqueue_orphaned_after: Duration, - queue: Queue, - ack: bool, -} - -impl Default for Config { - fn default() -> Self { - Self { - keep_alive: Duration::from_secs(30), - buffer_size: 10, - poll_strategy: StrategyBuilder::new() - .apply( - IntervalStrategy::new(Duration::from_millis(100)) - .with_backoff(BackoffConfig::default()), - ) - .build(), - reenqueue_orphaned_after: Duration::from_secs(300), // 5 minutes - queue: Queue::from("default"), - ack: true, - } - } -} - -impl Config { - /// Create a new config with a jobs queue - pub fn new(queue: &str) -> Self { - Config { - queue: Queue::from(queue), - ..Default::default() - } - } - - /// Interval between database poll queries - /// - /// Defaults to 100ms - pub fn with_poll_interval(mut self, strategy: MultiStrategy) -> Self { - self.poll_strategy = strategy; - self - } - - /// Interval between worker keep-alive database updates - /// - /// Defaults to 30s - pub fn set_keep_alive(mut self, keep_alive: Duration) -> Self { - self.keep_alive = keep_alive; - self - } - - /// Buffer size to use when querying for jobs - /// - /// Defaults to 10 - pub fn set_buffer_size(mut self, buffer_size: usize) -> Self { - self.buffer_size = buffer_size; - self - } - - /// Gets a reference to the keep_alive duration. - pub fn keep_alive(&self) -> &Duration { - &self.keep_alive - } - - /// Gets a mutable reference to the keep_alive duration. - pub fn keep_alive_mut(&mut self) -> &mut Duration { - &mut self.keep_alive - } - - /// Gets the buffer size. - pub fn buffer_size(&self) -> usize { - self.buffer_size - } - - /// Gets a reference to the poll_strategy. - pub fn poll_strategy(&self) -> &MultiStrategy { - &self.poll_strategy - } - - /// Gets a mutable reference to the poll_strategy. - pub fn poll_strategy_mut(&mut self) -> &mut MultiStrategy { - &mut self.poll_strategy - } +use crate::{CompactType, SqlContext, SqliteStorage}; - /// Gets a reference to the queue. - pub fn queue(&self) -> &Queue { - &self.queue - } - - /// Gets a mutable reference to the queue. - pub fn queue_mut(&mut self) -> &mut Queue { - &mut self.queue - } - - /// Gets the reenqueue_orphaned_after duration. - pub fn reenqueue_orphaned_after(&self) -> Duration { - self.reenqueue_orphaned_after - } - - /// Gets a mutable reference to the reenqueue_orphaned_after. - pub fn reenqueue_orphaned_after_mut(&mut self) -> &mut Duration { - &mut self.reenqueue_orphaned_after - } - - /// Occasionally some workers die, or abandon jobs because of panics. - /// This is the time a task takes before its back to the queue - /// - /// Defaults to 5 minutes - pub fn set_reenqueue_orphaned_after(mut self, after: Duration) -> Self { - self.reenqueue_orphaned_after = after; - self - } - - pub fn ack(&self) -> bool { - self.ack - } - - pub fn set_ack(mut self, auto_ack: bool) -> Self { - self.ack = auto_ack; - self - } -} +pub use apalis_sql::config::*; impl ConfigExt for SqliteStorage where SqliteStorage: - Backend, + Backend, { fn get_queue(&self) -> Queue { - self.config.queue.clone() + self.config().queue().clone() } } diff --git a/src/context.rs b/src/context.rs deleted file mode 100644 index 5e14694..0000000 --- a/src/context.rs +++ /dev/null @@ -1,152 +0,0 @@ -use std::{collections::HashMap, convert::Infallible}; - -use apalis_core::{task::metadata::MetadataExt, task_fn::FromRequest}; - -use serde::{ - Deserialize, Serialize, - de::{DeserializeOwned, Error}, -}; - -use crate::{CompactType, SqliteTask}; - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct SqliteContext { - max_attempts: i32, - last_result: Option, - lock_at: Option, - lock_by: Option, - done_at: Option, - priority: i32, - queue: Option, - meta: HashMap, -} - -impl Default for SqliteContext { - fn default() -> Self { - Self::new() - } -} - -impl SqliteContext { - /// Build a new context with defaults - pub fn new() -> Self { - SqliteContext { - lock_at: None, - done_at: None, - max_attempts: 5, - last_result: None, - lock_by: None, - priority: 0, - queue: None, - meta: HashMap::new(), - } - } - - /// Set the number of attempts - pub fn with_max_attempts(mut self, max_attempts: i32) -> Self { - self.max_attempts = max_attempts; - self - } - - /// Gets the maximum attempts for a job. Default 25 - pub fn max_attempts(&self) -> i32 { - self.max_attempts - } - - /// Get the time a job was done - pub fn done_at(&self) -> &Option { - &self.done_at - } - - /// Set the time a job was done - pub fn with_done_at(mut self, done_at: Option) -> Self { - self.done_at = done_at; - self - } - - /// Get the time a job was locked - pub fn lock_at(&self) -> &Option { - &self.lock_at - } - - /// Set the lock_at value - pub fn with_lock_at(mut self, lock_at: Option) -> Self { - self.lock_at = lock_at; - self - } - - /// Get the time a job was locked - pub fn lock_by(&self) -> &Option { - &self.lock_by - } - - /// Set `lock_by` - pub fn with_lock_by(mut self, lock_by: Option) -> Self { - self.lock_by = lock_by; - self - } - - /// Get the time a job was locked - pub fn last_result(&self) -> &Option { - &self.last_result - } - - /// Set the last result - pub fn with_last_result(mut self, result: Option) -> Self { - self.last_result = result; - self - } - - /// Set the job priority. Larger values will run sooner. Default is 0. - pub fn with_priority(mut self, priority: i32) -> Self { - self.priority = priority; - self - } - - /// Get the job priority - pub fn priority(&self) -> i32 { - self.priority - } - - pub fn queue(&self) -> &Option { - &self.queue - } - - pub fn with_queue(mut self, queue: String) -> Self { - self.queue = Some(queue); - self - } - - pub fn meta(&self) -> &HashMap { - &self.meta - } - - pub fn with_meta(mut self, meta: HashMap) -> Self { - self.meta = meta; - self - } -} - -impl FromRequest> for SqliteContext { - type Error = Infallible; - async fn from_request(req: &SqliteTask) -> Result { - Ok(req.parts.ctx.clone()) - } -} - -impl MetadataExt for SqliteContext { - type Error = serde_json::Error; - fn extract(&self) -> Result { - self.meta - .get(std::any::type_name::()) - .and_then(|v| serde_json::from_str::(v).ok()) - .ok_or(serde_json::Error::custom("Failed to extract metadata")) - } - fn inject(&mut self, value: T) -> Result<(), Self::Error> { - self.meta.insert( - std::any::type_name::().to_string(), - serde_json::to_string(&value).unwrap(), - ); - Ok(()) - } -} diff --git a/src/fetcher.rs b/src/fetcher.rs index 28e05d3..f754754 100644 --- a/src/fetcher.rs +++ b/src/fetcher.rs @@ -14,18 +14,19 @@ use apalis_core::{ task::Task, worker::context::WorkerContext, }; +use apalis_sql::{context::SqlContext, from_row::TaskRow}; use futures::{FutureExt, future::BoxFuture, stream::Stream}; use pin_project::pin_project; use sqlx::{Pool, Sqlite, SqlitePool}; use ulid::Ulid; -use crate::{CompactType, SqliteTask, config::Config, context::SqliteContext, from_row::TaskRow}; +use crate::{CompactType, SqliteTask, config::Config, from_row::SqliteTaskRow}; pub async fn fetch_next>( pool: SqlitePool, config: Config, worker: WorkerContext, -) -> Result>, sqlx::Error> +) -> Result>, sqlx::Error> where D::Error: std::error::Error + Send + Sync + 'static, Args: 'static, @@ -34,7 +35,7 @@ where let buffer_size = config.buffer_size() as i32; let worker = worker.name().to_string(); sqlx::query_file_as!( - TaskRow, + SqliteTaskRow, "queries/backend/fetch_next.sql", worker, job_type, @@ -43,7 +44,11 @@ where .fetch_all(&pool) .await? .into_iter() - .map(|r| r.try_into_task::()) + .map(|r| { + let row: TaskRow = r.try_into()?; + row.try_into_task::() + .map_err(|e| sqlx::Error::Protocol(e.to_string())) + }) .collect() } diff --git a/src/from_row.rs b/src/from_row.rs index 38d0489..8cd312f 100644 --- a/src/from_row.rs +++ b/src/from_row.rs @@ -4,8 +4,9 @@ use apalis_core::{ backend::codec::Codec, task::{attempt::Attempt, builder::TaskBuilder, status::Status, task_id::TaskId}, }; +use apalis_sql::context::SqlContext; -use crate::{CompactType, SqliteTask, context::SqliteContext}; +use crate::{CompactType, SqliteTask}; #[derive(Debug)] pub(crate) struct TaskRow { @@ -31,7 +32,7 @@ impl TaskRow { where D::Error: std::error::Error + Send + Sync + 'static, { - let ctx = SqliteContext::default() + let ctx = SqlContext::default() .with_done_at(self.done_at) .with_lock_by(self.lock_by) .with_max_attempts(self.max_attempts.unwrap_or(25) as i32) @@ -94,7 +95,7 @@ impl TaskRow { Ok(task.build()) } pub fn try_into_task_compact(self) -> Result, sqlx::Error> { - let ctx = SqliteContext::default() + let ctx = SqlContext::default() .with_done_at(self.done_at) .with_lock_by(self.lock_by) .with_max_attempts(self.max_attempts.unwrap_or(25) as i32) diff --git a/src/lib.rs b/src/lib.rs index 9b49277..6ddeb60 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -23,7 +23,7 @@ //! ### Basic Worker Example //! //! ```rust -//! # use apalis_sqlite::{SqliteStorage, SqliteContext}; +//! # use apalis_sqlite::{SqliteStorage, SqlContext}; //! # use apalis_core::task::Task; //! # use apalis_core::worker::context::WorkerContext; //! # use sqlx::SqlitePool; @@ -65,7 +65,7 @@ //! ### Hooked Worker Example (Event-driven) //! //! ```rust,no_run -//! # use apalis_sqlite::{SqliteStorage, SqliteContext, Config}; +//! # use apalis_sqlite::{SqliteStorage, SqlContext, Config}; //! # use apalis_core::task::Task; //! # use apalis_core::worker::context::WorkerContext; //! # use apalis_core::backend::poll_strategy::{IntervalStrategy, StrategyBuilder}; @@ -100,7 +100,7 @@ //! start += 1; //! Task::builder(serde_json::to_string(&start).unwrap()) //! .run_after(Duration::from_secs(1)) -//! .with_ctx(SqliteContext::new().with_priority(start)) +//! .with_ctx(SqlContext::new().with_priority(start)) //! .build() //! }) //! .take(20) @@ -128,7 +128,7 @@ //! ### Workflow Example //! //! ```rust,no_run -//! # use apalis_sqlite::{SqliteStorage, SqliteContext, Config}; +//! # use apalis_sqlite::{SqliteStorage, SqlContext, Config}; //! # use apalis_core::task::Task; //! # use apalis_core::worker::context::WorkerContext; //! # use sqlx::SqlitePool; @@ -200,7 +200,7 @@ //! ## License //! //! Licensed under either of Apache License, Version 2.0 or MIT license at your option. -//! +//! //! [`SqliteStorageWithHook`]: crate::SqliteStorage use std::{fmt, marker::PhantomData}; @@ -212,6 +212,7 @@ use apalis_core::{ task::Task, worker::{context::WorkerContext, ext::ack::AcknowledgeLayer}, }; +use apalis_sql::context::SqlContext; use futures::{ FutureExt, StreamExt, TryFutureExt, TryStreamExt, channel::mpsc, @@ -238,25 +239,88 @@ use crate::{ mod ack; mod callback; mod config; -mod context; pub mod fetcher; -pub mod from_row; pub mod queries; mod shared; pub mod sink; -pub type SqliteTask = Task; +mod from_row { + use chrono::{TimeZone, Utc}; + + #[derive(Debug)] + pub(crate) struct SqliteTaskRow { + pub(crate) job: Vec, + pub(crate) id: Option, + pub(crate) job_type: Option, + pub(crate) status: Option, + pub(crate) attempts: Option, + pub(crate) max_attempts: Option, + pub(crate) run_at: Option, + pub(crate) last_result: Option, + pub(crate) lock_at: Option, + pub(crate) lock_by: Option, + pub(crate) done_at: Option, + pub(crate) priority: Option, + pub(crate) metadata: Option, + } + + impl TryInto for SqliteTaskRow { + type Error = sqlx::Error; + + fn try_into(self) -> Result { + Ok(apalis_sql::from_row::TaskRow { + job: self.job, + id: self + .id + .ok_or_else(|| sqlx::Error::Protocol("Missing id".into()))?, + job_type: self + .job_type + .ok_or_else(|| sqlx::Error::Protocol("Missing job_type".into()))?, + status: self + .status + .ok_or_else(|| sqlx::Error::Protocol("Missing status".into()))?, + attempts: self + .attempts + .ok_or_else(|| sqlx::Error::Protocol("Missing attempts".into()))? + as usize, + max_attempts: self.max_attempts.map(|v| v as usize), + run_at: self.run_at.map(|ts| { + Utc.timestamp_opt(ts, 0) + .single() + .ok_or_else(|| sqlx::Error::Protocol("Invalid run_at timestamp".into())) + .unwrap() + }), + last_result: self + .last_result + .map(|res| serde_json::from_str(&res).unwrap_or(serde_json::Value::Null)), + lock_at: self.lock_at.map(|ts| { + Utc.timestamp_opt(ts, 0) + .single() + .ok_or_else(|| sqlx::Error::Protocol("Invalid run_at timestamp".into())) + .unwrap() + }), + lock_by: self.lock_by, + done_at: self.done_at.map(|ts| { + Utc.timestamp_opt(ts, 0) + .single() + .ok_or_else(|| sqlx::Error::Protocol("Invalid run_at timestamp".into())) + .unwrap() + }), + priority: self.priority.map(|v| v as usize), + metadata: self + .metadata + .map(|meta| serde_json::from_str(&meta).unwrap_or(serde_json::Value::Null)), + }) + } + } +} + +pub type SqliteTask = Task; pub use callback::{CallbackListener, DbEvent}; pub use config::Config; -pub use context::SqliteContext; pub use shared::{SharedPostgresError, SharedSqliteStorage}; pub use sqlx::SqlitePool; -#[cfg(feature = "json")] -pub type CompactType = String; - -// Bytes not yet supported due to sqlx limitations -#[cfg(feature = "bytes")] pub type CompactType = Vec; const INSERT_OPERATION: &str = "INSERT"; @@ -319,15 +383,7 @@ impl SqliteStorage<(), (), ()> { /// Get sqlite migrations without running them #[cfg(feature = "migrate")] pub fn migrations() -> sqlx::migrate::Migrator { - if cfg!(feature = "bytes") && cfg!(not(feature = "json")) { - sqlx::migrate!("./migrations/bytes") - } else if cfg!(feature = "json") && cfg!(not(feature = "bytes")) { - sqlx::migrate!("./migrations/json") - } else { - panic!( - "One of either the 'json' or 'bytes' feature must be enabled for migrations to work." - ); - } + sqlx::migrate!("./migrations") } } @@ -454,7 +510,7 @@ where type Args = Args; type IdType = Ulid; - type Context = SqliteContext; + type Context = SqlContext; type Codec = Decode; @@ -515,7 +571,7 @@ where type Args = Args; type IdType = Ulid; - type Context = SqliteContext; + type Context = SqlContext; type Codec = Decode; @@ -678,9 +734,9 @@ mod tests { let items = stream::repeat_with(move || { start += 1; - Task::builder(serde_json::to_string(&start).unwrap()) + Task::builder(serde_json::to_vec(&start).unwrap()) .run_after(Duration::from_secs(1)) - .with_ctx(SqliteContext::new().with_priority(start)) + .with_ctx(SqlContext::new().with_priority(start)) .build() }) .take(ITEMS) @@ -824,6 +880,7 @@ mod tests { } }) .then(|a: Vec, mut worker: WorkerContext| async move { + dbg!(&a); worker.emit(&Event::Custom(Box::new(format!( "Generated {} summaries", a.len() diff --git a/src/queries/fetch_by_id.rs b/src/queries/fetch_by_id.rs index 4b8aad6..939770d 100644 --- a/src/queries/fetch_by_id.rs +++ b/src/queries/fetch_by_id.rs @@ -2,14 +2,15 @@ use apalis_core::{ backend::{Backend, FetchById, codec::Codec}, task::task_id::TaskId, }; +use apalis_sql::from_row::{FromRowError, TaskRow}; use ulid::Ulid; -use crate::{CompactType, SqliteContext, SqliteStorage, SqliteTask, from_row::TaskRow}; +use crate::{CompactType, SqlContext, SqliteStorage, SqliteTask, from_row::SqliteTaskRow}; impl FetchById for SqliteStorage where SqliteStorage: - Backend, + Backend, D: Codec, D::Error: std::error::Error + Send + Sync + 'static, Args: 'static, @@ -21,11 +22,17 @@ where let pool = self.pool.clone(); let id = id.to_string(); async move { - let task = sqlx::query_file_as!(TaskRow, "queries/task/find_by_id.sql", id) + let task = sqlx::query_file_as!(SqliteTaskRow, "queries/task/find_by_id.sql", id) .fetch_optional(&pool) .await? - .map(|r| r.try_into_task::()) - .transpose()?; + .map(|r| { + let row: TaskRow = r + .try_into() + .map_err(|e: sqlx::Error| FromRowError::DecodeError(e.into()))?; + row.try_into_task::() + }) + .transpose() + .map_err(|e| sqlx::Error::Protocol(e.to_string()))?; Ok(task) } } diff --git a/src/queries/list_queues.rs b/src/queries/list_queues.rs index 2be5499..698ee46 100644 --- a/src/queries/list_queues.rs +++ b/src/queries/list_queues.rs @@ -1,7 +1,7 @@ use apalis_core::backend::{Backend, ListQueues, QueueInfo}; use ulid::Ulid; -use crate::{CompactType, SqliteContext, SqliteStorage}; +use crate::{CompactType, SqlContext, SqliteStorage}; struct QueueInfoRow { name: String, @@ -24,7 +24,7 @@ impl From for QueueInfo { impl ListQueues for SqliteStorage where SqliteStorage: - Backend, + Backend, { fn list_queues(&self) -> impl Future, Self::Error>> + Send { let pool = self.pool.clone(); diff --git a/src/queries/list_tasks.rs b/src/queries/list_tasks.rs index e4d5a75..3a36561 100644 --- a/src/queries/list_tasks.rs +++ b/src/queries/list_tasks.rs @@ -2,14 +2,15 @@ use apalis_core::{ backend::{Backend, Filter, ListAllTasks, ListTasks, codec::Codec}, task::{Task, status::Status}, }; +use apalis_sql::from_row::TaskRow; use ulid::Ulid; -use crate::{CompactType, SqliteContext, SqliteStorage, SqliteTask, from_row::TaskRow}; +use crate::{CompactType, SqlContext, SqliteStorage, SqliteTask, from_row::SqliteTaskRow}; impl ListTasks for SqliteStorage where SqliteStorage: - Backend, + Backend, D: Codec, D::Error: std::error::Error + Send + Sync + 'static, Args: 'static, @@ -30,7 +31,7 @@ where .to_string(); async move { let tasks = sqlx::query_file_as!( - TaskRow, + SqliteTaskRow, "queries/backend/list_jobs.sql", status, queue, @@ -40,7 +41,11 @@ where .fetch_all(&pool) .await? .into_iter() - .map(|r| r.try_into_task::()) + .map(|r| { + let row: TaskRow = r.try_into()?; + row.try_into_task::() + .map_err(|e| sqlx::Error::Protocol(e.to_string())) + }) .collect::, _>>()?; Ok(tasks) } @@ -50,7 +55,7 @@ where impl ListAllTasks for SqliteStorage where SqliteStorage: - Backend, + Backend, { fn list_all_tasks( &self, @@ -68,7 +73,7 @@ where let offset = filter.offset() as i32; async move { let tasks = sqlx::query_file_as!( - TaskRow, + SqliteTaskRow, "queries/backend/list_all_jobs.sql", status, limit, @@ -77,7 +82,11 @@ where .fetch_all(&pool) .await? .into_iter() - .map(|r| r.try_into_task_compact()) + .map(|r| { + let row: TaskRow = r.try_into()?; + row.try_into_task_compact() + .map_err(|e| sqlx::Error::Protocol(e.to_string())) + }) .collect::, _>>()?; Ok(tasks) } diff --git a/src/queries/list_workers.rs b/src/queries/list_workers.rs index 9dcedec..4e7e34b 100644 --- a/src/queries/list_workers.rs +++ b/src/queries/list_workers.rs @@ -2,7 +2,7 @@ use apalis_core::backend::{Backend, ListWorkers, RunningWorker}; use futures::TryFutureExt; use ulid::Ulid; -use crate::{CompactType, SqliteContext, SqliteStorage}; +use crate::{CompactType, SqlContext, SqliteStorage}; struct Worker { id: String, @@ -16,7 +16,7 @@ struct Worker { impl ListWorkers for SqliteStorage where SqliteStorage: - Backend, + Backend, { fn list_workers( &self, diff --git a/src/queries/metrics.rs b/src/queries/metrics.rs index ec31493..f5df479 100644 --- a/src/queries/metrics.rs +++ b/src/queries/metrics.rs @@ -1,7 +1,7 @@ use apalis_core::backend::{Backend, Metrics, Statistic}; use ulid::Ulid; -use crate::{CompactType, SqliteContext, SqliteStorage}; +use crate::{CompactType, SqlContext, SqliteStorage}; struct StatisticRow { /// The priority of the statistic (lower number means higher priority) @@ -17,7 +17,7 @@ struct StatisticRow { impl Metrics for SqliteStorage where SqliteStorage: - Backend, + Backend, { fn global(&self) -> impl Future, Self::Error>> + Send { let pool = self.pool.clone(); diff --git a/src/queries/mod.rs b/src/queries/mod.rs index 64dfd8f..e79ace6 100644 --- a/src/queries/mod.rs +++ b/src/queries/mod.rs @@ -9,6 +9,7 @@ pub mod metrics; pub mod reenqueue_orphaned; pub mod register_worker; pub mod wait_for; +pub mod vacuum; fn stat_type_from_string(s: &str) -> StatType { match s { diff --git a/src/queries/vacuum.rs b/src/queries/vacuum.rs new file mode 100644 index 0000000..d1c3275 --- /dev/null +++ b/src/queries/vacuum.rs @@ -0,0 +1,20 @@ +use apalis_core::backend::{Backend, Vacuum}; +use ulid::Ulid; + +use crate::{CompactType, SqliteStorage}; + +impl Vacuum for SqliteStorage +where + SqliteStorage: + Backend, + F: Send, + Decode: Send, + Args: Send, +{ + async fn vacuum(&mut self) -> Result { + let res = sqlx::query_file!("queries/backend/vacuum.sql") + .execute(&self.pool) + .await?; + Ok(res.rows_affected() as usize) + } +} diff --git a/src/queries/wait_for.rs b/src/queries/wait_for.rs index ddaf54d..3626f57 100644 --- a/src/queries/wait_for.rs +++ b/src/queries/wait_for.rs @@ -14,7 +14,7 @@ use crate::{CompactType, SqliteStorage}; struct ResultRow { pub id: Option, pub status: Option, - pub result: Option, + pub result: Option, } impl WaitForCompletion for SqliteStorage diff --git a/src/shared.rs b/src/shared.rs index 719d5d1..f7aee62 100644 --- a/src/shared.rs +++ b/src/shared.rs @@ -13,11 +13,10 @@ use crate::{ CompactType, Config, INSERT_OPERATION, JOBS_TABLE, SqliteStorage, SqliteTask, ack::{LockTaskLayer, SqliteAck}, callback::{DbEvent, update_hook_callback}, - context::SqliteContext, fetcher::SqlitePollFetcher, initial_heartbeat, keep_alive, }; -use crate::{from_row::TaskRow, sink::SqliteSink}; +use crate::{from_row::SqliteTaskRow, sink::SqliteSink}; use apalis_core::{ backend::{ Backend, TaskStream, @@ -26,6 +25,7 @@ use apalis_core::{ }, worker::{context::WorkerContext, ext::ack::AcknowledgeLayer}, }; +use apalis_sql::{context::SqlContext, from_row::TaskRow}; use futures::{ FutureExt, SinkExt, Stream, StreamExt, TryStreamExt, channel::mpsc::{self, Receiver, Sender}, @@ -90,14 +90,18 @@ impl SharedSqliteStorage> { let mut tx = pool.begin().await?; let buffer_size = max(10, instances.len()) as i32; let res: Vec<_> = sqlx::query_file_as!( - TaskRow, + SqliteTaskRow, "queries/backend/fetch_next_shared.sql", job_types, row_ids, buffer_size, ) .fetch(&mut *tx) - .map(|r| r?.try_into_task_compact()) + .map(|r| { + let row: TaskRow = r?.try_into()?; + row.try_into_task_compact() + .map_err(|e| sqlx::Error::Protocol(e.to_string())) + }) .try_collect() .await?; tx.commit().await?; @@ -221,7 +225,7 @@ where type Compact = CompactType; - type Context = SqliteContext; + type Context = SqlContext; type Layer = Stack>; From aad3cd76627c240f8db69fe0985cb2b9ad67001e Mon Sep 17 00:00:00 2001 From: geofmureithi Date: Sat, 25 Oct 2025 18:51:09 +0300 Subject: [PATCH 3/3] bump: to new version --- Cargo.toml | 17 +++++------------ README.md | 17 +++++------------ src/ack.rs | 22 +++------------------- src/lib.rs | 2 +- src/shared.rs | 2 +- 5 files changed, 15 insertions(+), 45 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 2522107..bb7e249 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "apalis-sqlite" -version = "1.0.0-alpha.2" +version = "1.0.0-alpha.3" authors = ["Njuguna Mureithi "] readme = "README.md" edition = "2024" @@ -25,11 +25,8 @@ features = ["chrono", "sqlite"] [dependencies] serde = { version = "1", features = ["derive"] } serde_json = { version = "1" } -apalis-core = { version = "1.0.0-alpha.6", default-features = false, features = [ - "sleep", -], path = "../apalis/packages/apalis-core" } -apalis-sql = { version = "1.0.0-alpha.6", path = "../apalis/packages/apalis-sql" } -apalis-workflow = { version = "0.1.0-alpha.4", path = "../apalis/packages/apalis-workflow" } +apalis-core = { version = "1.0.0-alpha.7", features = ["sleep"] } +apalis-sql = { version = "1.0.0-alpha.7" } log = "0.4.21" futures = "0.3.30" tokio = { version = "1", features = ["rt", "net"], optional = true } @@ -39,17 +36,13 @@ thiserror = "2.0.0" pin-project = "1.1.10" libsqlite3-sys = "0.30.1" ulid = { version = "1", features = ["serde"] } -tower-layer = "0.3.3" -tower-service = "0.3.3" bytes = "1.1.0" [dev-dependencies] tokio = { version = "1", features = ["macros", "rt-multi-thread"] } -apalis-core = { version = "1.0.0-alpha.6", features = [ - "test-utils", -], path = "../apalis/packages/apalis-core" } -apalis-workflow = { version = "0.1.0-alpha.4", path = "../apalis/packages/apalis-workflow" } +apalis-core = { version = "1.0.0-alpha.7", features = ["test-utils"] } apalis-sqlite = { path = ".", features = ["migrate", "tokio-comp"] } +apalis-workflow = { version = "0.1.0-alpha.5" } [package.metadata.docs.rs] # defines the configuration attribute `docsrs` diff --git a/README.md b/README.md index ca9a3c3..3f14fd2 100644 --- a/README.md +++ b/README.md @@ -1,12 +1,12 @@ # apalis-sqlite -Background task processing for Rust using Apalis and SQLite. +Background task processing for Rust using apalis and sqlite. ## Features - **Reliable job queue** using SQLite as the backend. - **Multiple storage types**: standard polling and event-driven (hooked) storage. -- **Custom codecs** for serializing/deserializing job arguments as bytes and json. +- **Custom codecs** for serializing/deserializing job arguments as bytes. - **Heartbeat and orphaned job re-enqueueing** for robust job processing. - **Integration with Apalis workers and middleware.** @@ -144,17 +144,10 @@ async fn main() { } ``` -## Migrations +## Observability -If the `migrate` feature is enabled, you can run built-in migrations with: - -```rust,no_run -use sqlx::SqlitePool; -#[tokio::main] async fn main() { - let pool = SqlitePool::connect(":memory:").await.unwrap(); - apalis_sqlite::SqliteStorage::setup(&pool).await.unwrap(); -} -``` +You can track your jobs using [apalis-board](https://github.com/apalis-dev/apalis-board). +![Task](https://github.com/apalis-dev/apalis-board/raw/master/screenshots/task.png) ## License diff --git a/src/ack.rs b/src/ack.rs index e96c1c5..e226223 100644 --- a/src/ack.rs +++ b/src/ack.rs @@ -1,21 +1,16 @@ -use std::any::Any; - use apalis_core::{ error::{AbortError, BoxDynError}, + layers::{Layer, Service}, task::{Parts, status::Status}, worker::{context::WorkerContext, ext::ack::Acknowledge}, }; use apalis_sql::context::SqlContext; -use apalis_workflow::StepResult; use futures::{FutureExt, future::BoxFuture}; use serde::Serialize; -use serde_json::Value; use sqlx::SqlitePool; -use tower_layer::Layer; -use tower_service::Service; use ulid::Ulid; -use crate::{CompactType, SqliteTask}; +use crate::SqliteTask; #[derive(Clone)] pub struct SqliteAck { @@ -39,18 +34,7 @@ impl Acknowledge for SqliteAck let worker_id = parts.ctx.lock_by().clone(); // Workflows need special handling to serialize the response correctly - let response = match res { - Ok(r) => { - if let Some(res_ref) = (r as &dyn Any).downcast_ref::>() { - let res_deserialized: Result = - serde_json::from_slice(&res_ref.0); - serde_json::to_string(&res_deserialized.map_err(|e| e.to_string())) - } else { - serde_json::to_string(&res.as_ref().map_err(|e| e.to_string())) - } - } - _ => serde_json::to_string(&res.as_ref().map_err(|e| e.to_string())), - }; + let response = serde_json::to_string(&res.as_ref().map_err(|e| e.to_string())); let status = calculate_status(parts, res); parts.status.store(status.clone()); diff --git a/src/lib.rs b/src/lib.rs index 6ddeb60..063537f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -211,6 +211,7 @@ use apalis_core::{ }, task::Task, worker::{context::WorkerContext, ext::ack::AcknowledgeLayer}, + layers::Stack, }; use apalis_sql::context::SqlContext; use futures::{ @@ -222,7 +223,6 @@ use futures::{ use libsqlite3_sys::{sqlite3, sqlite3_update_hook}; use sqlx::{Pool, Sqlite}; use std::ffi::c_void; -use tower_layer::Stack; use ulid::Ulid; use crate::{ diff --git a/src/shared.rs b/src/shared.rs index f7aee62..c18764e 100644 --- a/src/shared.rs +++ b/src/shared.rs @@ -24,6 +24,7 @@ use apalis_core::{ shared::MakeShared, }, worker::{context::WorkerContext, ext::ack::AcknowledgeLayer}, + layers::Stack, }; use apalis_sql::{context::SqlContext, from_row::TaskRow}; use futures::{ @@ -35,7 +36,6 @@ use futures::{ }; use libsqlite3_sys::{sqlite3, sqlite3_update_hook}; use sqlx::SqlitePool; -use tower_layer::Stack; use ulid::Ulid; pub struct SharedSqliteStorage {