From 9591e8fc68f70de7eced580c0af523ed814a8017 Mon Sep 17 00:00:00 2001 From: geofmureithi Date: Sat, 10 Oct 2026 00:49:31 +0300 Subject: [PATCH] bump: to v1.0.0-rc.12 --- .github/workflows/audit.yaml | 63 +++--- CHANGELOG.md | 1 + Cargo.lock | 34 ++-- apalis-core/Cargo.toml | 2 +- apalis-core/README.md | 20 +- apalis-core/src/backend/ext/delegate.rs | 95 +++++++++ apalis-core/src/backend/ext/inspect_err.rs | 2 +- apalis-core/src/backend/ext/instrument.rs | 2 +- apalis-core/src/backend/ext/interleave.rs | 2 +- .../src/backend/ext/lifecycle/after_start.rs | 1 + .../src/backend/ext/lifecycle/after_stop.rs | 1 + .../src/backend/ext/lifecycle/before_start.rs | 1 + .../src/backend/ext/lifecycle/before_stop.rs | 1 + apalis-core/src/backend/ext/map_err.rs | 2 +- .../backend/ext/poll_strategy/poll_with.rs | 1 + apalis-core/src/backend/ext/wake_on_push.rs | 2 +- apalis-core/src/backend/ext/with_codec.rs | 6 +- apalis-core/src/backend/impls/mod.rs | 191 +++++++++++++++++- apalis-core/src/worker/builder.rs | 2 +- apalis-core/src/worker/mod.rs | 2 +- apalis-workflow/Cargo.toml | 6 +- apalis/Cargo.toml | 6 +- apalis/README.md | 6 +- examples/advanced/src/main.rs | 5 +- supply-chain/config.toml | 18 +- utils/apalis-codec/Cargo.toml | 4 +- utils/apalis-file-storage/Cargo.toml | 6 +- 27 files changed, 389 insertions(+), 93 deletions(-) diff --git a/.github/workflows/audit.yaml b/.github/workflows/audit.yaml index b1c3b574..c2180cc0 100644 --- a/.github/workflows/audit.yaml +++ b/.github/workflows/audit.yaml @@ -1,21 +1,20 @@ name: Security audit on: schedule: - - cron: '0 0 * * *' + - cron: "0 0 * * *" push: branches: [main, develop] - paths: - - '**/Cargo.toml' - - '**/Cargo.lock' - - '**/deny.toml' + paths: + - "**/Cargo.toml" + - "**/Cargo.lock" + - "**/deny.toml" pull_request: - paths: - - '**/Cargo.toml' - - '**/Cargo.lock' - - '**/deny.toml' + paths: + - "**/Cargo.toml" + - "**/Cargo.lock" + - "**/deny.toml" workflow_call: - jobs: cargo-audit: @@ -29,30 +28,36 @@ jobs: key: cargo-audit-${{ runner.os }} - run: cargo install cargo-audit --locked --force - run: cargo audit --ignore RUSTSEC-2024-0436 #(used in ratatui) - + cargo-vet: runs-on: ubuntu-latest env: CARGO_VET_VERSION: 0.10.1 steps: - - uses: actions/checkout@master - - name: Install Rust - run: rustup update stable && rustup default stable - - uses: actions/cache@v6 - with: - path: ${{ runner.tool_cache }}/cargo-vet - key: cargo-vet-bin-${{ env.CARGO_VET_VERSION }} - - name: Add the tool cache directory to the search path - run: echo "${{ runner.tool_cache }}/cargo-vet/bin" >> $GITHUB_PATH - - name: Ensure that the tool cache is populated with the cargo-vet binary - run: cargo install --root ${{ runner.tool_cache }}/cargo-vet --version ${{ env.CARGO_VET_VERSION }} cargo-vet - - name: Generate Lockfile - run: cargo generate-lockfile - - name: Invoke cargo-vet - run: cargo vet --locked + - uses: actions/checkout@master + - name: Install Rust + run: rustup update stable && rustup default stable + - uses: actions/cache@v6 + with: + path: ${{ runner.tool_cache }}/cargo-vet + key: cargo-vet-bin-${{ env.CARGO_VET_VERSION }} + - name: Add the tool cache directory to the search path + run: echo "${{ runner.tool_cache }}/cargo-vet/bin" >> $GITHUB_PATH + - name: Ensure that the tool cache is populated with the cargo-vet binary + run: cargo install --root ${{ runner.tool_cache }}/cargo-vet --version ${{ env.CARGO_VET_VERSION }} cargo-vet + - name: Generate Lockfile + run: cargo generate-lockfile + - name: Invoke cargo-vet + run: cargo vet --locked cargo-deny: - runs-on: ubuntu-22.04 + runs-on: ubuntu-latest steps: - - uses: actions/checkout@v6 - - uses: EmbarkStudios/cargo-deny-action@v2 + - uses: actions/checkout@master + - uses: dtolnay/rust-toolchain@stable + - uses: actions/cache@v6 + with: + path: ~/.cargo/bin/cargo-deny + key: cargo-deny-${{ runner.os }} + - run: cargo install cargo-deny --locked + - run: cargo deny check diff --git a/CHANGELOG.md b/CHANGELOG.md index 2dadeb1a..59906085 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ All notable changes to this project are documented in this file. ## [Unreleased] +- *bump*: to v1.0.0-rc.12 ([#791](https://github.com/apalis-dev/apalis/pull/791)) - *bump*: to v1.0.0-rc.11 ([#790](https://github.com/apalis-dev/apalis/pull/790)) - **fix(core)**: wake idle workers when monitor shutdown starts ([#777](https://github.com/apalis-dev/apalis/pull/777)) - *bump*: to v1.0.0-rc.10 ([#775](https://github.com/apalis-dev/apalis/pull/775)) diff --git a/Cargo.lock b/Cargo.lock index c2c3b9a1..7109e779 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -202,7 +202,7 @@ checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" [[package]] name = "apalis" -version = "1.0.0-rc.11" +version = "1.0.0-rc.12" dependencies = [ "apalis-core", "apalis-file-storage", @@ -229,7 +229,7 @@ dependencies = [ [[package]] name = "apalis-codec" -version = "0.1.0-rc.11" +version = "0.1.0-rc.12" dependencies = [ "apalis-core", "bincode", @@ -241,7 +241,7 @@ dependencies = [ [[package]] name = "apalis-core" -version = "1.0.0-rc.11" +version = "1.0.0-rc.12" dependencies = [ "dashmap", "document-features", @@ -265,7 +265,7 @@ dependencies = [ [[package]] name = "apalis-file-storage" -version = "0.1.0-rc.11" +version = "0.1.0-rc.12" dependencies = [ "apalis-codec", "apalis-core", @@ -284,7 +284,7 @@ dependencies = [ [[package]] name = "apalis-workflow" -version = "0.1.0-rc.11" +version = "0.1.0-rc.12" dependencies = [ "apalis-codec", "apalis-core", @@ -1968,9 +1968,9 @@ dependencies = [ [[package]] name = "opentelemetry" -version = "0.33.0" +version = "0.33.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6cdb0b1b267eb9db3331b434ed9ddab10d50e280a9adf9d13e5233e2002b61b5" +checksum = "8c1f1a6ac9c875a4941e4c7c085b61e8f4daa326de03b4701768c3491a6ad9ee" dependencies = [ "futures-core", "futures-sink", @@ -2002,9 +2002,9 @@ dependencies = [ [[package]] name = "opentelemetry-prometheus" -version = "0.33.0" +version = "0.33.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "274c2e3a2ad55524484500d5a78dffbfee23e4a1e4c3d3fcbd881b61b033e6df" +checksum = "b9b52a5722eb64d198d6b6f1189e6b75a3c5ddfd224f48ede00c4330e0f11c66" dependencies = [ "once_cell", "opentelemetry", @@ -2014,9 +2014,9 @@ dependencies = [ [[package]] name = "opentelemetry_sdk" -version = "0.33.0" +version = "0.33.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cb39533d9d1c912123efd7d41d7e0c29d16917b60ce15b4c8d87cb1af7f67520" +checksum = "372a912db2974d1990207006fd0e3aa0ef3fd4ad3be43c65cfaadf2206866061" dependencies = [ "futures-channel", "futures-executor", @@ -3074,9 +3074,9 @@ dependencies = [ [[package]] name = "tokio-util" -version = "0.7.19" +version = "0.7.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" +checksum = "e464cf451ba96ebfc6f9b6542f17ee8b8956e33f1e40d9690624e59d7a7f8a4b" dependencies = [ "bytes", "futures-core", @@ -3694,18 +3694,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.61" +version = "0.8.62" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "879fb705ce98c32e41ebdb970fbe1204f8492423b314c6ab0354c3e7b5542866" +checksum = "86502bf56ac7c77571a32e2647bb2a15894565e981fb2a48d7bde2d91c965a9d" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.61" +version = "0.8.62" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "708882a28301d604fa039cc7727607a98d04c4b86dc76ec9cc683f805d709759" +checksum = "5457206954b06561e2608c7e19cf58b1926586d999c246eebe4502f7e2039d1a" dependencies = [ "proc-macro2", "quote", diff --git a/apalis-core/Cargo.toml b/apalis-core/Cargo.toml index 684ec02a..421ae481 100644 --- a/apalis-core/Cargo.toml +++ b/apalis-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "apalis-core" -version = "1.0.0-rc.11" +version = "1.0.0-rc.12" authors = ["Njuguna Mureithi "] edition.workspace = true repository.workspace = true diff --git a/apalis-core/README.md b/apalis-core/README.md index 7d6dba8f..c1571777 100644 --- a/apalis-core/README.md +++ b/apalis-core/README.md @@ -48,11 +48,11 @@ Specific documentation for tasks can be found in the [`task`] and [`task::builde ##### Relevant Guides: -- [**Defining Task arguments**](https://docs.rs/apalis-core/1.0.0-rc.11/apalis_core/task_fn/guide/index.html) - Creating effective task arguments that are scalable and type-safe +- [**Defining Task arguments**](https://docs.rs/apalis-core/1.0.0-rc.12/apalis_core/task_fn/guide/index.html) - Creating effective task arguments that are scalable and type-safe ### Backends -The [`Backend`](https://docs.rs/apalis-core/1.0.0-rc.11/apalis_core/backend/trait.Backend.html) trait serves as the core abstraction for all task sources. +The [`Backend`](https://docs.rs/apalis-core/1.0.0-rc.12/apalis_core/backend/trait.Backend.html) trait serves as the core abstraction for all task sources. It defines task polling mechanisms, streaming interfaces, and middleware integration points.
@@ -68,16 +68,16 @@ It defines task polling mechanisms, streaming interfaces, and middleware integra #### Inbuilt Implementations -- [`MemoryStorage`](https://docs.rs/apalis-core/1.0.0-rc.11/apalis_core/backend/memory/struct.MemoryStorage.html) : In-memory storage based on channels -- [`Pipe`](https://docs.rs/apalis-core/1.0.0-rc.11/apalis_core/backend/pipe/index.html) : Pipe-based backend for a stream-to-backend pipeline -- [`CustomBackend`](https://docs.rs/apalis-core/1.0.0-rc.11/apalis_core/backend/custom/index.html) : Flexible backend composition allowing custom functions for task management +- [`MemoryStorage`](https://docs.rs/apalis-core/1.0.0-rc.12/apalis_core/backend/memory/struct.MemoryStorage.html) : In-memory storage based on channels +- [`Pipe`](https://docs.rs/apalis-core/1.0.0-rc.12/apalis_core/backend/pipe/index.html) : Pipe-based backend for a stream-to-backend pipeline +- [`CustomBackend`](https://docs.rs/apalis-core/1.0.0-rc.12/apalis_core/backend/custom/index.html) : Flexible backend composition allowing custom functions for task management Backends handle task persistence, distribution, and reliability concerns while providing a uniform interface for worker consumption. ### Workers -The [`Worker`](https://docs.rs/apalis-core/1.0.0-rc.11/apalis_core/worker/index.html) is the core runtime component responsible for task polling, execution, and lifecycle management: +The [`Worker`](https://docs.rs/apalis-core/1.0.0-rc.12/apalis_core/worker/index.html) is the core runtime component responsible for task polling, execution, and lifecycle management: #### Worker Lifecycle @@ -94,7 +94,7 @@ The following are the main components the worker module: - [`Worker`] - Actual worker implementation that processes tasks - [`WorkerContext`] - Runtime state including task counts and execution status - [`Event`] - Worker event enumeration (`Start`, `Engage`, `Idle`, `Error`, `Stop`) -- [`Ext`](https://docs.rs/apalis-core/1.0.0-rc.11/apalis_core/worker/ext/index.html) - Extension traits and middleware for adding functionality to workers +- [`Ext`](https://docs.rs/apalis-core/1.0.0-rc.12/apalis_core/worker/ext/index.html) - Extension traits and middleware for adding functionality to workers #### Example: Building and Running a Worker @@ -129,11 +129,11 @@ Learn more about workers in the [`worker`](crate::worker) and [`worker::builder` ##### Relevant Tutorials: - [**Creating task handlers**](crate::task::task_fn) - Defining task processing functions using the [`TaskFn`] trait -- [**Testing task handlers with `TestWorker`**](https://docs.rs/apalis-core/1.0.0-rc.11/apalis_core/worker/test_worker/index.html) - Specialized worker implementation for unit and integration testing +- [**Testing task handlers with `TestWorker`**](https://docs.rs/apalis-core/1.0.0-rc.12/apalis_core/worker/test_worker/index.html) - Specialized worker implementation for unit and integration testing ### Monitor -The [`Monitor`](https://docs.rs/apalis-core/1.0.0-rc.11/apalis_core/monitor/struct.Monitor.html) helps manage and coordinate multiple workers: +The [`Monitor`](https://docs.rs/apalis-core/1.0.0-rc.12/apalis_core/monitor/struct.Monitor.html) helps manage and coordinate multiple workers: **Main Features:** @@ -166,7 +166,7 @@ async fn main() { } ``` -Learn more about the monitor in the [`monitor` module](https://docs.rs/apalis-core/1.0.0-rc.11/apalis_core/monitor/index.html. +Learn more about the monitor in the [`monitor` module](https://docs.rs/apalis-core/1.0.0-rc.12/apalis_core/monitor/index.html. ### Middleware diff --git a/apalis-core/src/backend/ext/delegate.rs b/apalis-core/src/backend/ext/delegate.rs index f2d2d339..47195960 100644 --- a/apalis-core/src/backend/ext/delegate.rs +++ b/apalis-core/src/backend/ext/delegate.rs @@ -317,5 +317,100 @@ macro_rules! delegate_expose { $wrap_body } } + + impl<$($lt,)* $($generic),+> ListQueues for $wrapper + where + B: ListQueues + Send + Sync, + $($bound)+ + { + async fn list_queues( + &self, + ) -> Result, Self::Error> { + let $result = self.$field.list_queues().await; + #[allow(unused_variables)] + let $this = &*self; + $wrap_body + } + } + + impl<$($lt,)* $($generic),+> ListWorkers for $wrapper + where + B: ListWorkers + Send + Sync, + $($bound)+ + { + async fn list_workers( + &self, + ) -> Result, Self::Error> { + let $result = self.$field.list_workers().await; + #[allow(unused_variables)] + let $this = &*self; + $wrap_body + } + + async fn list_all_workers( + &self, + ) -> Result, Self::Error> { + let $result = self.$field.list_all_workers().await; + #[allow(unused_variables)] + let $this = &*self; + $wrap_body + } + } + + impl<$($lt,)* $($generic),+> ListTasks for $wrapper + where + B: ListTasks + Send + Sync, + $($bound)+ + { + async fn list_tasks( + &self, + filter: &Filter, + ) -> Result>, Self::Error> { + let $result = self.$field.list_tasks(filter).await; + #[allow(unused_variables)] + let $this = &*self; + $wrap_body + } + } + + impl<$($lt,)* $($generic),+> ListAllTasks for $wrapper + where + B: ListAllTasks + Send + Sync, + $($bound)+ + { + async fn list_all_tasks( + &self, + filter: &Filter, + ) -> Result>, Self::Error> { + let $result = self.$field.list_all_tasks(filter).await; + #[allow(unused_variables)] + let $this = &*self; + $wrap_body + } + } + + impl<$($lt,)* $($generic),+> Metrics for $wrapper + where + B: Metrics + Send + Sync, + $($bound)+ + { + async fn global( + &self, + ) -> Result, Self::Error> { + let $result = self.$field.global().await; + #[allow(unused_variables)] + let $this = &*self; + $wrap_body + } + + async fn fetch_by_queue( + &self, + ) -> Result, Self::Error> { + let $result = self.$field.fetch_by_queue().await; + #[allow(unused_variables)] + let $this = &*self; + $wrap_body + } + } }; } diff --git a/apalis-core/src/backend/ext/inspect_err.rs b/apalis-core/src/backend/ext/inspect_err.rs index b7b47bec..5eec0b2c 100644 --- a/apalis-core/src/backend/ext/inspect_err.rs +++ b/apalis-core/src/backend/ext/inspect_err.rs @@ -7,7 +7,7 @@ use futures_sink::Sink; use futures_util::SinkExt; use futures_util::TryStreamExt; -use crate::{backend::*, worker::context::WorkerContext}; +use crate::{backend::*, task::Task, worker::context::WorkerContext}; /// A `Backend` wrapper that runs a callback `F` on each error yielded by the poll stream. #[derive(Debug, Clone)] diff --git a/apalis-core/src/backend/ext/instrument.rs b/apalis-core/src/backend/ext/instrument.rs index 07bc631a..98e61762 100644 --- a/apalis-core/src/backend/ext/instrument.rs +++ b/apalis-core/src/backend/ext/instrument.rs @@ -5,7 +5,7 @@ use std::task::{Context, Poll}; use tracing::Span; -use crate::{backend::*, worker::context::WorkerContext}; +use crate::{backend::*, task::Task, worker::context::WorkerContext}; /// Instruments the inner [Backend] with the provided `Span`, returning an /// `Instrumented` wrapper. diff --git a/apalis-core/src/backend/ext/interleave.rs b/apalis-core/src/backend/ext/interleave.rs index 29a68062..d7db517a 100644 --- a/apalis-core/src/backend/ext/interleave.rs +++ b/apalis-core/src/backend/ext/interleave.rs @@ -7,7 +7,7 @@ use futures_core::Stream; use futures_sink::Sink; use futures_util::SinkExt; -use crate::{backend::*, worker::context::WorkerContext}; +use crate::{backend::*, task::Task, worker::context::WorkerContext}; /// A `Backend` wrapper that interleaves tasks from an external/shared `Stream` /// with tasks produced by the wrapped backend. diff --git a/apalis-core/src/backend/ext/lifecycle/after_start.rs b/apalis-core/src/backend/ext/lifecycle/after_start.rs index fa5f01fb..4b90ec5f 100644 --- a/apalis-core/src/backend/ext/lifecycle/after_start.rs +++ b/apalis-core/src/backend/ext/lifecycle/after_start.rs @@ -17,6 +17,7 @@ use crate::{ future::BoxSyncFuture, *, }, + task::Task, worker::context::WorkerContext, }; diff --git a/apalis-core/src/backend/ext/lifecycle/after_stop.rs b/apalis-core/src/backend/ext/lifecycle/after_stop.rs index 04502692..7f0db320 100644 --- a/apalis-core/src/backend/ext/lifecycle/after_stop.rs +++ b/apalis-core/src/backend/ext/lifecycle/after_stop.rs @@ -18,6 +18,7 @@ use crate::{ future::BoxSyncFuture, *, }, + task::Task, worker::context::WorkerContext, }; diff --git a/apalis-core/src/backend/ext/lifecycle/before_start.rs b/apalis-core/src/backend/ext/lifecycle/before_start.rs index 91a15b1a..eab0c5d9 100644 --- a/apalis-core/src/backend/ext/lifecycle/before_start.rs +++ b/apalis-core/src/backend/ext/lifecycle/before_start.rs @@ -18,6 +18,7 @@ use crate::{ future::BoxSyncFuture, *, }, + task::Task, worker::context::WorkerContext, }; diff --git a/apalis-core/src/backend/ext/lifecycle/before_stop.rs b/apalis-core/src/backend/ext/lifecycle/before_stop.rs index bd07cf0f..424dde3d 100644 --- a/apalis-core/src/backend/ext/lifecycle/before_stop.rs +++ b/apalis-core/src/backend/ext/lifecycle/before_stop.rs @@ -17,6 +17,7 @@ use crate::{ future::BoxSyncFuture, *, }, + task::Task, worker::context::WorkerContext, }; diff --git a/apalis-core/src/backend/ext/map_err.rs b/apalis-core/src/backend/ext/map_err.rs index 36a56e3b..efe57599 100644 --- a/apalis-core/src/backend/ext/map_err.rs +++ b/apalis-core/src/backend/ext/map_err.rs @@ -7,7 +7,7 @@ use futures_sink::Sink; use futures_util::SinkExt; use futures_util::TryStreamExt; -use crate::{backend::*, worker::context::WorkerContext}; +use crate::{backend::*, task::Task, worker::context::WorkerContext}; /// A `Backend` wrapper that maps the backend's error type `Self::Error` into `E2`. #[derive(Debug, Clone)] diff --git a/apalis-core/src/backend/ext/poll_strategy/poll_with.rs b/apalis-core/src/backend/ext/poll_strategy/poll_with.rs index b12715fc..1b7774ae 100644 --- a/apalis-core/src/backend/ext/poll_strategy/poll_with.rs +++ b/apalis-core/src/backend/ext/poll_strategy/poll_with.rs @@ -14,6 +14,7 @@ use crate::{ ext::poll_strategy::{PollMetrics, PollStrategy}, *, }, + task::Task, worker::context::WorkerContext, }; diff --git a/apalis-core/src/backend/ext/wake_on_push.rs b/apalis-core/src/backend/ext/wake_on_push.rs index 383f77c1..47c92ca9 100644 --- a/apalis-core/src/backend/ext/wake_on_push.rs +++ b/apalis-core/src/backend/ext/wake_on_push.rs @@ -6,7 +6,7 @@ use std::{ use futures_sink::Sink; use futures_util::SinkExt; -use crate::{backend::*, worker::context::WorkerContext}; +use crate::{backend::*, task::Task, worker::context::WorkerContext}; /// A backend wrapper that wakes the worker when a task is pushed. /// diff --git a/apalis-core/src/backend/ext/with_codec.rs b/apalis-core/src/backend/ext/with_codec.rs index 1fd78d81..190d1d99 100644 --- a/apalis-core/src/backend/ext/with_codec.rs +++ b/apalis-core/src/backend/ext/with_codec.rs @@ -9,7 +9,7 @@ use std::{ use futures_sink::Sink; use futures_util::SinkExt; -use crate::{backend::*, worker::context::WorkerContext}; +use crate::{backend::*, task::Task, worker::context::WorkerContext}; /// A `Backend` wrapper that swaps out the serialization codec entirely (JSON, /// MessagePack, Protobuf, ...) without touching storage logic. @@ -94,9 +94,9 @@ delegate_config!(WithCodec, backend); delegate_expose!( impl for WithCodec where { - B: Send + Sync + Backend, + B: Send + Sync + Backend + BackendConfig + WireFormatBackend, C: Send + Sync + 'static, - + C: Codec + Send + 'static, } => backend ); diff --git a/apalis-core/src/backend/impls/mod.rs b/apalis-core/src/backend/impls/mod.rs index 389d01ac..393070c4 100644 --- a/apalis-core/src/backend/impls/mod.rs +++ b/apalis-core/src/backend/impls/mod.rs @@ -1,11 +1,19 @@ -use std::task::{Context, Poll}; +use std::{ + task::{Context, Poll}, + time::Duration, +}; use futures_core::stream::BoxStream; use futures_util::StreamExt; use tower_layer::Identity; use crate::{ - backend::{Backend, BackendConfig, finalize::Ephemeral}, + backend::{ + Backend, BackendConfig, FetchById, Filter, ListAllTasks, ListQueues, ListTasks, + ListWorkers, Metrics, QueueInfo, RegisterWorker, Reschedule, ResumeAbandoned, ResumeById, + RunningWorker, Statistic, TaskResult, Update, Vacuum, WaitForCompletion, WireFormatBackend, + finalize::Ephemeral, + }, task::{ Task, builder::TaskBuilder, @@ -108,3 +116,182 @@ impl Backend for &mut B { (**self).poll_close(cx, worker) } } + +impl BackendConfig for &mut B { + type Args = B::Args; + type Kind = B::Kind; + + type Id = B::Id; + + type Config = B::Config; + + type Layer = B::Layer; + + fn config(&self) -> &Self::Config { + BackendConfig::config(*self) + } + + fn middleware(&mut self, worker: &mut WorkerContext) -> Self::Layer { + BackendConfig::middleware(*self, worker) + } +} + +impl WireFormatBackend for &mut B { + type Codec = B::Codec; + + type Compact = B::Compact; + + fn codec(&self) -> &Self::Codec { + WireFormatBackend::codec(*self) + } +} + +impl FetchById for &mut B +where + B: FetchById, + B::Task: Send, + B: Backend + Send, +{ + async fn fetch_by_id( + &mut self, + task_id: &crate::task::task_id::TaskId, + ) -> Result, Self::Error> { + ::fetch_by_id(*self, task_id).await + } +} +impl Update for &mut B +where + B: Update + Send, + B::Task: Send, + B: Backend + Send, +{ + async fn update(&mut self, task: Self::Task) -> Result<(), Self::Error> { + ::update(*self, task).await + } +} +impl Reschedule for &mut B +where + B: Reschedule, + B::Task: Send, + B: Backend + Send, +{ + async fn reschedule( + &mut self, + task: Self::Task, + wait: std::time::Duration, + ) -> Result<(), Self::Error> { + ::reschedule(*self, task, wait).await + } +} +impl Vacuum for &mut B +where + B: Vacuum, + B: Backend + Send, +{ + async fn vacuum(&mut self) -> Result { + ::vacuum(*self).await + } + async fn vacuum_before(&mut self, duration: Duration) -> Result { + ::vacuum_before(*self, duration).await + } +} +impl ResumeById for &mut B +where + B: ResumeById, + B: Backend + Send, +{ + async fn resume_by_id(&mut self, id: TaskId) -> Result { + ::resume_by_id(*self, id).await + } +} +impl ResumeAbandoned for &mut B +where + B: ResumeAbandoned, + B: Backend + Send, +{ + async fn resume_abandoned(&mut self) -> Result { + ::resume_abandoned(*self).await + } +} +impl RegisterWorker for &mut B +where + B: RegisterWorker, + B: Backend + Send, +{ + async fn register_worker(&mut self, worker_id: String) -> Result<(), Self::Error> { + ::register_worker(*self, worker_id).await + } +} +impl WaitForCompletion for &mut B +where + B: WaitForCompletion + Sync, + Output: 'static, + B: Backend + Send, +{ + type ResultStream = BoxStream<'static, Result, Self::Error>>; + fn wait_for(&mut self, task_ids: impl IntoIterator) -> Self::ResultStream { + use futures_util::StreamExt; + let result = >::wait_for(*self, task_ids); + result.boxed() + } + async fn check_status( + &mut self, + task_ids: impl IntoIterator + Send, + ) -> Result>, Self::Error> { + >::check_status(*self, task_ids).await + } +} +impl ListQueues for &mut B +where + B: ListQueues + Send + Sync, + B: Backend + Send, +{ + async fn list_queues(&self) -> Result, Self::Error> { + ::list_queues(*self).await + } +} +impl ListWorkers for &mut B +where + B: ListWorkers + Send + Sync, + B: Backend + Send, +{ + async fn list_workers(&self) -> Result, Self::Error> { + ::list_workers(*self).await + } + async fn list_all_workers(&self) -> Result, Self::Error> { + ::list_all_workers(*self).await + } +} +impl ListTasks for &mut B +where + B: ListTasks + Send + Sync, + B: Backend + Send, +{ + async fn list_tasks(&self, filter: &Filter) -> Result>, Self::Error> { + ::list_tasks(*self, filter).await + } +} +impl ListAllTasks for &mut B +where + B: ListAllTasks + Send + Sync, + B: Backend + Send, +{ + async fn list_all_tasks( + &self, + filter: &Filter, + ) -> Result>, Self::Error> { + ::list_all_tasks(*self, filter).await + } +} +impl Metrics for &mut B +where + B: Metrics + Send + Sync, + B: Backend + Send, +{ + async fn global(&self) -> Result, Self::Error> { + ::global(*self).await + } + async fn fetch_by_queue(&self) -> Result, Self::Error> { + ::fetch_by_queue(*self).await + } +} diff --git a/apalis-core/src/worker/builder.rs b/apalis-core/src/worker/builder.rs index 454206dc..040fe24e 100644 --- a/apalis-core/src/worker/builder.rs +++ b/apalis-core/src/worker/builder.rs @@ -215,7 +215,7 @@ impl WorkerBuilder { where W: IntoWorkerService, B: Backend + BackendConfig, - NB: Backend + BackendConfig + Send + Unpin + 'static, + NB: Backend + BackendConfig + Send + Unpin, Svc: Service, Args: Send + 'static, { diff --git a/apalis-core/src/worker/mod.rs b/apalis-core/src/worker/mod.rs index c42d6580..89439c64 100644 --- a/apalis-core/src/worker/mod.rs +++ b/apalis-core/src/worker/mod.rs @@ -182,7 +182,7 @@ impl Worker { impl Worker where - FB: BackendConfig + Backend> + Send + Unpin + 'static, + FB: BackendConfig + Backend> + Send + Unpin, S: Service> + Send + 'static, FB::Args: Send + 'static, Args: Send + 'static, diff --git a/apalis-workflow/Cargo.toml b/apalis-workflow/Cargo.toml index f0785479..d08950de 100644 --- a/apalis-workflow/Cargo.toml +++ b/apalis-workflow/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "apalis-workflow" -version = "0.1.0-rc.11" +version = "0.1.0-rc.12" edition.workspace = true repository.workspace = true authors = ["Njuguna Mureithi "] @@ -18,12 +18,12 @@ petgraph = { version = "0.8.3", features = ["serde-1"] } serde = { version = "1.0", features = ["derive"] } serde_json = "1" tower = { version = "0.5", features = ["util"], default-features = false } -apalis-core = { path = "../apalis-core", version = "1.0.0-rc.11", default-features = false, features = [ +apalis-core = { path = "../apalis-core", version = "1.0.0-rc.12", default-features = false, features = [ "sleep", "serde", "shared" ] } -apalis-codec = { path = "../utils/apalis-codec", version = "0.1.0-rc.11" } +apalis-codec = { path = "../utils/apalis-codec", version = "0.1.0-rc.12" } futures-util = { version = "0.3.34", features = [ "sink", "async-await", diff --git a/apalis/Cargo.toml b/apalis/Cargo.toml index da102fd3..dd0d6739 100644 --- a/apalis/Cargo.toml +++ b/apalis/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "apalis" -version = "1.0.0-rc.11" +version = "1.0.0-rc.12" authors = ["Geoffrey Mureithi "] description = "Simple, extensible and multithreaded background task processing for Rust" edition.workspace = true @@ -78,7 +78,7 @@ full = [ docsrs = ["document-features"] [dependencies.apalis-core] -version = "1.0.0-rc.11" +version = "1.0.0-rc.12" default-features = false path = "../apalis-core" @@ -110,7 +110,7 @@ version = "0.1.44" optional = true [dev-dependencies] -apalis-core = { path = "../apalis-core", version = "1.0.0-rc.11" } +apalis-core = { path = "../apalis-core", version = "1.0.0-rc.12" } serde = "1" tokio = { version = "1", features = ["full"] } tracing-subscriber = { version = "0.3.23", default-features = false, features = [ diff --git a/apalis/README.md b/apalis/README.md index d0a672df..f871768e 100644 --- a/apalis/README.md +++ b/apalis/README.md @@ -32,7 +32,7 @@ ## Features -- **Simple, predictable task handling** - [Task handlers](https://docs.rs/apalis-core/1.0.0-rc.11/apalis_core/task/task_fn/index.html) are just async functions, with a macro-free API +- **Simple, predictable task handling** - [Task handlers](https://docs.rs/apalis-core/1.0.0-rc.12/apalis_core/task/task_fn/index.html) are just async functions, with a macro-free API - **Robust task execution** - Built-in retries, rate limiting, timeouts, idempotency, and error handling - **Multiple storage backends** - Redis, PostgreSQL, SQLite, and in-memory storage - **Advanced task management** - Prioritization, scheduling, metadata, and result tracking @@ -65,8 +65,8 @@ To get started, just add to Cargo.toml ```toml [dependencies] -apalis = { version = "1.0.0-rc.11" } -# apalis-redis = { version = "1.0.0-rc.11" } # Use redis/sqlite/postgres etc +apalis = { version = "1.0.0-rc.12" } +# apalis-redis = { version = "1.0.0-rc.12" } # Use redis/sqlite/postgres etc ``` ## Usage diff --git a/examples/advanced/src/main.rs b/examples/advanced/src/main.rs index 0bbf85b4..058747bb 100644 --- a/examples/advanced/src/main.rs +++ b/examples/advanced/src/main.rs @@ -38,6 +38,7 @@ async fn send_email( ctx: TaskContext, ) -> Result<(), Error> { let address = email.to.clone(); + let c = ctx.clone(); // Fast path: avoid the expensive validation request when the address // is already known to be valid. @@ -67,6 +68,8 @@ async fn send_email( .instrument(Span::current()), ); + c.cancel().unwrap(); + Ok(()) } @@ -89,7 +92,7 @@ async fn main() -> Result<(), BoxDynError> { produce_jobs(&mut backend).await; WorkerBuilder::new("email-sender") - .backend(backend) + .backend(&mut backend) // Spawn each tasks future via tokio::spawn .parallelize(tokio::spawn) // Process several emails concurrently. diff --git a/supply-chain/config.toml b/supply-chain/config.toml index da08eb35..deaacfde 100644 --- a/supply-chain/config.toml +++ b/supply-chain/config.toml @@ -66,15 +66,15 @@ version = "1.0.104" criteria = "safe-to-deploy" [[exemptions.apalis]] -version = "1.0.0-rc.11" +version = "1.0.0-rc.12" criteria = "safe-to-deploy" [[exemptions.apalis-core]] -version = "1.0.0-rc.11" +version = "1.0.0-rc.12" criteria = "safe-to-deploy" [[exemptions.apalis-workflow]] -version = "0.1.0-rc.11" +version = "0.1.0-rc.12" criteria = "safe-to-deploy" [[exemptions.atomic-waker]] @@ -694,15 +694,15 @@ version = "0.9.117" criteria = "safe-to-deploy" [[exemptions.opentelemetry]] -version = "0.33.0" +version = "0.33.1" criteria = "safe-to-deploy" [[exemptions.opentelemetry-prometheus]] -version = "0.33.0" +version = "0.33.1" criteria = "safe-to-deploy" [[exemptions.opentelemetry_sdk]] -version = "0.33.0" +version = "0.33.1" criteria = "safe-to-deploy" [[exemptions.os_info]] @@ -1090,7 +1090,7 @@ version = "0.26.6" criteria = "safe-to-deploy" [[exemptions.tokio-util]] -version = "0.7.19" +version = "0.7.20" criteria = "safe-to-deploy" [[exemptions.tower]] @@ -1334,11 +1334,11 @@ version = "0.8.4" criteria = "safe-to-deploy" [[exemptions.zerocopy]] -version = "0.8.61" +version = "0.8.62" criteria = "safe-to-deploy" [[exemptions.zerocopy-derive]] -version = "0.8.61" +version = "0.8.62" criteria = "safe-to-deploy" [[exemptions.zerofrom]] diff --git a/utils/apalis-codec/Cargo.toml b/utils/apalis-codec/Cargo.toml index 10dbd45f..89eb9e57 100644 --- a/utils/apalis-codec/Cargo.toml +++ b/utils/apalis-codec/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "apalis-codec" -version = "0.1.0-rc.11" +version = "0.1.0-rc.12" rust-version.workspace = true edition.workspace = true repository.workspace = true @@ -21,7 +21,7 @@ bincode = ["dep:bincode", "dep:thiserror"] msgpack = ["dep:rmp-serde", "dep:serde", "apalis-core/serde", "dep:thiserror"] [dependencies] -apalis-core = { path = "../../apalis-core", version = "1.0.0-rc.11", default-features = false } +apalis-core = { path = "../../apalis-core", version = "1.0.0-rc.12", default-features = false } serde = { version = "1.0", features = ["derive"], optional = true } serde_json = { version = "1", optional = true } bincode = { version = "=2.0.1", optional = true } diff --git a/utils/apalis-file-storage/Cargo.toml b/utils/apalis-file-storage/Cargo.toml index 25981091..22885421 100644 --- a/utils/apalis-file-storage/Cargo.toml +++ b/utils/apalis-file-storage/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "apalis-file-storage" -version = "0.1.0-rc.11" +version = "0.1.0-rc.12" rust-version.workspace = true edition.workspace = true repository.workspace = true @@ -16,7 +16,7 @@ categories = ["database", "filesystem", "asynchronous", "data-structures"] [dependencies] serde = { version = "1.0.229", features = ["derive"] } -apalis-core = { path = "../../apalis-core", version = "1.0.0-rc.11", default-features = false, features = [ +apalis-core = { path = "../../apalis-core", version = "1.0.0-rc.12", default-features = false, features = [ "sleep", "serde", ] } @@ -33,7 +33,7 @@ futures-util = { version = "0.3.34", features = [ "std", ], default-features = false } futures-core = { version = "0.3.34", default-features = false } -apalis-codec = { path = "../apalis-codec", version = "0.1.0-rc.11", default-features = false, features = [ +apalis-codec = { path = "../apalis-codec", version = "0.1.0-rc.12", default-features = false, features = [ "json", ] } tempfile = "3.27.0"