apalis_diesel_postgres/
lib.rs

1#![doc = include_str!("../README.md")]
2#![warn(missing_docs)]
3#![warn(rustdoc::broken_intra_doc_links)]
4
5use std::{fmt::Debug, marker::PhantomData};
6
7pub use apalis_codec::json::JsonCodec;
8use apalis_core::{
9    backend::{Backend, BackendExt, TaskStream, codec::Codec, queue::Queue},
10    task::{Task, task_id::TaskId},
11    worker::context::WorkerContext,
12};
13pub use apalis_sql::{config::Config, from_row::TaskRow};
14use diesel::{
15    PgConnection,
16    r2d2::{ConnectionManager, Pool},
17};
18use futures::{StreamExt, TryStreamExt};
19use ulid::Ulid;
20
21pub use crate::{
22    ack::{PgAck, PgMiddleware, lock_task, lock_task_in_queue},
23    error::Error,
24    fetcher::{PgFetcher, PgNotify},
25    lifecycle::{refresh_queue_stats_snapshot, setup, verify_schema},
26    pool::{build_pool, build_pool_with},
27    queries::migrations::MIGRATIONS,
28    shared::{SharedFetcher, SharedPostgresError, SharedPostgresStorage},
29    sink::PgSink,
30};
31
32mod ack;
33mod admin;
34mod error;
35mod fetcher;
36mod lifecycle;
37mod models;
38mod notify_event;
39mod pool;
40mod queries;
41mod runtime;
42mod shared;
43mod sink;
44
45pub(crate) use notify_event::InsertEvent;
46pub mod schema;
47
48/// Diesel/r2d2 PostgreSQL pool accepted by this backend.
49pub type PgPool = Pool<ConnectionManager<PgConnection>>;
50/// SQL context associated with PostgreSQL tasks.
51pub type PgContext = apalis_sql::context::SqlContext<PgPool>;
52/// Apalis task stored in PostgreSQL.
53pub type PgTask<Args> = Task<Args, PgContext, Ulid>;
54/// PostgreSQL task identifier.
55pub type PgTaskId = TaskId<Ulid>;
56/// Compact serialized payload representation.
57pub type CompactType = Vec<u8>;
58
59/// Canonical `apalis.workers.storage_name` value for this backend. Shared by
60/// the worker registration path and the admin `RegisterWorker` UPSERT so they
61/// cannot drift apart.
62pub(crate) const STORAGE_NAME: &str = "PostgresStorage";
63
64/// Returns the crate name.
65#[must_use]
66pub const fn crate_name() -> &'static str {
67    "apalis-diesel-postgres"
68}
69
70// apalis `WorkerBuilder::build()` requires the backend to be `Send + Sync`.
71const _: fn() = || {
72    const fn assert_send_sync<T: Send + Sync>() {}
73    assert_send_sync::<PostgresStorage<()>>();
74    assert_send_sync::<PostgresStorage<(), JsonCodec<CompactType>, PgNotify>>();
75    assert_send_sync::<PostgresStorage<(), JsonCodec<CompactType>, SharedFetcher>>();
76    assert_send_sync::<SharedPostgresStorage<()>>();
77};
78
79/// PostgreSQL storage backend implemented with Diesel.
80pub struct PostgresStorage<
81    Args,
82    Codec = JsonCodec<CompactType>,
83    Fetcher = PgFetcher<CompactType, Codec>,
84> {
85    _marker: PhantomData<(Args, Codec)>,
86    pub(crate) pool: PgPool,
87    pub(crate) config: Config,
88    pub(crate) fetcher: Fetcher,
89    pub(crate) sink: PgSink<Args, Codec>,
90    /// Per-process lease token. Generated at construction (and shared by
91    /// clones via `Arc`) so that the `keep_alive` heartbeat can only be
92    /// refreshed by code holding this storage handle — even if a third party
93    /// learns the `(worker_id, queue)` pair, they cannot extend the heartbeat
94    /// without also possessing the token.
95    pub(crate) lease_token: std::sync::Arc<str>,
96}
97
98// Manual Unpin requires Fetcher: Unpin so pinning guarantees from a
99// `!Unpin` fetcher (e.g. one holding a self-referential future) are
100// honoured by the storage wrapper. All built-in fetchers (PgFetcher,
101// PgNotify, SharedFetcher) satisfy this bound.
102impl<Args, Codec, Fetcher: Unpin> Unpin for PostgresStorage<Args, Codec, Fetcher> {}
103
104impl<Args, Codec, Fetcher: Debug> Debug for PostgresStorage<Args, Codec, Fetcher> {
105    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
106        f.debug_struct("PostgresStorage")
107            .field("config", &self.config)
108            .field("fetcher", &self.fetcher)
109            .finish_non_exhaustive()
110    }
111}
112
113impl<Args, Codec, Fetcher: Clone> Clone for PostgresStorage<Args, Codec, Fetcher> {
114    fn clone(&self) -> Self {
115        Self {
116            _marker: PhantomData,
117            pool: self.pool.clone(),
118            config: self.config.clone(),
119            fetcher: self.fetcher.clone(),
120            sink: self.sink.clone(),
121            lease_token: self.lease_token.clone(),
122        }
123    }
124}
125
126impl<Args> PostgresStorage<Args> {
127    /// Create storage for the queue named after `Args`.
128    ///
129    /// **Do not share `pool` with HTTP request handlers or other unrelated
130    /// workloads.** Apalis holds long-lived connections (fetcher, lifecycle
131    /// keep-alive, listener) and a backend that exhausts a shared pool will
132    /// stall the worker, causing heartbeat loss and orphan reenqueue
133    /// cascades. See the README section "Connection pool isolation" for the
134    /// recommended sizing and the [`Self::push_with_conn`] outbox API for the
135    /// supported way to enqueue from a backend transaction.
136    #[must_use]
137    pub fn new(pool: &PgPool) -> Self {
138        let config = Config::new(std::any::type_name::<Args>());
139        Self::new_with_config(pool, &config)
140    }
141
142    /// Create storage with an explicit Apalis SQL config.
143    ///
144    /// **Do not share `pool` with HTTP request handlers or other unrelated
145    /// workloads** — see [`Self::new`] for the rationale.
146    #[must_use]
147    pub fn new_with_config(pool: &PgPool, config: &Config) -> Self {
148        Self {
149            _marker: PhantomData,
150            pool: pool.clone(),
151            config: config.clone(),
152            fetcher: PgFetcher::default(),
153            sink: PgSink::new(pool, config),
154            lease_token: queries::worker::mint_lease_token().into(),
155        }
156    }
157
158    /// Create storage that also listens for PostgreSQL notifications.
159    ///
160    /// Notify mode uses a dedicated pooled connection for `LISTEN
161    /// "apalis::job::insert"` while the polling stream is alive. **Each
162    /// `new_with_notify` storage spawns one listener thread and pins one
163    /// pool connection.** If you need notify-driven dequeue across many
164    /// queues, prefer [`crate::SharedPostgresStorage`] — it spawns a single
165    /// listener thread shared by all queues registered with it, so the
166    /// thread/connection cost stays at one regardless of queue count.
167    ///
168    /// **Do not share `pool` with HTTP request handlers or other unrelated
169    /// workloads** — see [`Self::new`] for the rationale.
170    #[must_use]
171    pub fn new_with_notify(
172        pool: &PgPool,
173        config: &Config,
174    ) -> PostgresStorage<Args, JsonCodec<CompactType>, PgNotify> {
175        PostgresStorage {
176            _marker: PhantomData,
177            pool: pool.clone(),
178            config: config.clone(),
179            fetcher: PgNotify,
180            sink: PgSink::new(pool, config),
181            lease_token: queries::worker::mint_lease_token().into(),
182        }
183    }
184
185    /// Return the underlying Diesel/r2d2 pool.
186    #[must_use]
187    pub fn pool(&self) -> &PgPool {
188        &self.pool
189    }
190
191    /// Return the queue configuration.
192    #[must_use]
193    pub fn config(&self) -> &Config {
194        &self.config
195    }
196}
197
198impl<Args, Codec, Fetcher> PostgresStorage<Args, Codec, Fetcher> {
199    /// Change the task codec while retaining pool, config, fetcher, and the
200    /// sink's pipeline state (buffered tasks and any in-flight flush — the
201    /// buffer holds already-encoded compact tasks, so switching the codec
202    /// must not silently drop them).
203    #[must_use]
204    pub fn with_codec<NewCodec>(self) -> PostgresStorage<Args, NewCodec, Fetcher> {
205        PostgresStorage {
206            _marker: PhantomData,
207            sink: self.sink.retype(),
208            pool: self.pool,
209            config: self.config,
210            fetcher: self.fetcher,
211            lease_token: self.lease_token,
212        }
213    }
214
215    /// Compose the keep-alive + reenqueue heartbeat stream shared by every
216    /// `Backend` impl on this storage. Kept here so the per-fetcher
217    /// `Backend::heartbeat` impls remain one-liners.
218    pub(crate) fn heartbeat_stream(
219        &self,
220        worker: &WorkerContext,
221    ) -> futures::stream::BoxStream<'static, Result<(), Error>> {
222        let keep_alive = queries::keep_alive_stream(
223            self.pool.clone(),
224            self.config.clone(),
225            worker.clone(),
226            std::sync::Arc::clone(&self.lease_token),
227        );
228        let reenqueue = queries::reenqueue_orphaned_stream(self.pool.clone(), self.config.clone())
229            .map_ok(|_| ());
230        futures::stream::select(keep_alive, reenqueue).boxed()
231    }
232}
233
234/// Transactional enqueue on a caller-supplied connection — the **outbox**
235/// entry point. Use these methods when you want the task INSERT to share a
236/// transaction with your business-data writes.
237impl<Args, EncodeCodec, Fetcher> PostgresStorage<Args, EncodeCodec, Fetcher>
238where
239    EncodeCodec: Codec<Args, Compact = CompactType>,
240    EncodeCodec::Error: std::error::Error + Send + Sync + 'static,
241{
242    /// Enqueue a task using a caller-supplied `PgConnection`.
243    ///
244    /// For transactional outbox semantics, call this inside
245    /// `conn.transaction(|c| ...)` together with your business-data writes —
246    /// the INSERT into `apalis.jobs` is committed only if the outer
247    /// transaction commits. Without an outer transaction, Diesel auto-commits
248    /// the INSERT (same behaviour as the pool-based `Sink<Task>` path).
249    ///
250    /// NOTIFY is delivered when the (outer) transaction commits, so listeners
251    /// only see tasks that were actually committed. No manual `pg_notify` is
252    /// needed.
253    ///
254    /// This is the only synchronous public method in the crate. From an async
255    /// context, invoke it on the selected runtime's blocking pool —
256    /// `tokio::task::spawn_blocking` with the `tokio` feature, or
257    /// `ntex_rt::spawn_blocking` with the `ntex` feature — together with your
258    /// business-data writes so the entire transaction lives on one blocking
259    /// task.
260    ///
261    /// See [`Self::push_task_with_conn`] for the full-control variant that
262    /// accepts a pre-built [`PgTask`] (custom `idempotency_key`, `priority`,
263    /// `run_at`, `max_attempts`, `metadata`, or `task_id`).
264    ///
265    /// # Errors
266    /// - [`Error::Decode`] if the codec rejects `args`.
267    /// - [`Error::InvalidArgument`] if serialized metadata exceeds the byte
268    ///   cap, or for unreachable `run_at`.
269    /// - [`Error::Database`] for SQL/driver failures.
270    pub fn push_with_conn(&self, conn: &mut PgConnection, args: Args) -> Result<PgTaskId, Error> {
271        let encoded = EncodeCodec::encode(&args).map_err(|err| Error::Decode(Box::new(err)))?;
272        let task_id = PgTaskId::new(Ulid::new());
273        let mut task = PgTask::<CompactType>::new(encoded);
274        task.parts.task_id = Some(task_id);
275        queries::push_tasks_on_conn(conn, &self.config, vec![task])?;
276        Ok(task_id)
277    }
278
279    /// Enqueue a fully-constructed [`PgTask<Args>`] using a caller-supplied
280    /// connection. Use this when you need to set `idempotency_key`,
281    /// `priority`, `run_at`, `max_attempts`, `metadata`, or a specific
282    /// `task_id`.
283    ///
284    /// Semantics are identical to [`Self::push_with_conn`]; see that method's
285    /// docs for the transaction/NOTIFY contract.
286    ///
287    /// If `task.parts.task_id` is `None`, a fresh Ulid is generated and
288    /// returned. If `Some`, that id is used as-is and echoed back.
289    ///
290    /// # Errors
291    /// - [`Error::Decode`] if the codec rejects the task's `args`.
292    /// - [`Error::IdempotencyConflict`] on `idempotency_key` conflict — the
293    ///   savepoint for this batch is rolled back (the whole batch, not just the
294    ///   duplicate), but your outer transaction continues; decide whether to
295    ///   commit or roll back based on the error.
296    /// - [`Error::InvalidArgument`] if serialized metadata exceeds the byte
297    ///   cap.
298    /// - [`Error::Database`] for SQL/driver failures.
299    pub fn push_task_with_conn(
300        &self,
301        conn: &mut PgConnection,
302        task: PgTask<Args>,
303    ) -> Result<PgTaskId, Error> {
304        let encoded =
305            EncodeCodec::encode(&task.args).map_err(|err| Error::Decode(Box::new(err)))?;
306        let task_id = task
307            .parts
308            .task_id
309            .unwrap_or_else(|| PgTaskId::new(Ulid::new()));
310        let mut compact = PgTask::<CompactType> {
311            args: encoded,
312            parts: task.parts,
313        };
314        compact.parts.task_id = Some(task_id);
315        queries::push_tasks_on_conn(conn, &self.config, vec![compact])?;
316        Ok(task_id)
317    }
318}
319
320/// Single generic `Backend` impl covering every `Fetcher: PgFetcherSource`.
321/// Heartbeat/middleware are identical for all three modes; the per-mode
322/// pipeline is delegated through `PgFetcherSource::into_compact_stream`.
323impl<Args, Decode, Fetcher> Backend for PostgresStorage<Args, Decode, Fetcher>
324where
325    Args: Send + 'static + Unpin,
326    Decode: Codec<Args, Compact = CompactType> + Send + 'static,
327    Decode::Error: std::error::Error + Send + Sync + 'static,
328    Fetcher: crate::fetcher::PgFetcherSource,
329{
330    type Args = Args;
331    type IdType = Ulid;
332    type Context = PgContext;
333    type Error = Error;
334    type Stream = TaskStream<PgTask<Args>, Error>;
335    type Beat = futures::stream::BoxStream<'static, Result<(), Error>>;
336    type Layer = PgMiddleware;
337
338    fn heartbeat(&self, worker: &WorkerContext) -> Self::Beat {
339        self.heartbeat_stream(worker)
340    }
341
342    fn middleware(&self) -> Self::Layer {
343        PgMiddleware::with_lease_token(
344            self.pool.clone(),
345            self.config.ack(),
346            std::sync::Arc::clone(&self.lease_token),
347        )
348    }
349
350    fn poll(self, worker: &WorkerContext) -> Self::Stream {
351        // The decode stage keeps its own pool handle (and the claiming
352        // worker's id) so it can fail rows whose payload does not decode —
353        // see `queries::fail_undecodable_task`.
354        let pool = self.pool.clone();
355        let compact = self.fetcher.into_compact_stream(
356            self.pool,
357            self.config,
358            worker.clone(),
359            self.lease_token,
360        );
361        crate::fetcher::decode_task_stream::<Args, Decode>(compact, pool, worker.name().to_owned())
362    }
363}
364
365impl<Args, Decode, Fetcher> BackendExt for PostgresStorage<Args, Decode, Fetcher>
366where
367    Args: Send + 'static + Unpin,
368    Decode: Codec<Args, Compact = CompactType> + Send + 'static,
369    Decode::Error: std::error::Error + Send + Sync + 'static,
370    Fetcher: crate::fetcher::PgFetcherSource,
371{
372    type Compact = CompactType;
373    type Codec = Decode;
374    type CompactStream = TaskStream<PgTask<CompactType>, Self::Error>;
375
376    fn get_queue(&self) -> Queue {
377        self.config.queue().clone()
378    }
379
380    fn poll_compact(self, worker: &WorkerContext) -> Self::CompactStream {
381        self.fetcher
382            .into_compact_stream(self.pool, self.config, worker.clone(), self.lease_token)
383    }
384}
385
386#[cfg(test)]
387mod tests {
388    use apalis_core::{
389        backend::{Backend, BackendExt},
390        task::status::Status,
391    };
392    use apalis_sql::{DateTime, DateTimeExt, from_row::FromRowError};
393    use diesel::{
394        PgConnection,
395        r2d2::{ConnectionManager, Pool},
396    };
397    use lets_expect::{AssertionError, AssertionResult, *};
398
399    use super::*;
400
401    fn row(
402        id: &str,
403        status: &str,
404        run_at: Option<DateTime>,
405        idempotency_key: Option<&str>,
406    ) -> TaskRow {
407        TaskRow {
408            job: b"payload".to_vec(),
409            id: id.to_owned(),
410            job_type: "unit-queue".to_owned(),
411            status: status.to_owned(),
412            attempts: 2,
413            max_attempts: Some(3),
414            run_at,
415            last_result: None,
416            lock_at: None,
417            lock_by: Some("worker-a".to_owned()),
418            done_at: None,
419            priority: Some(7),
420            metadata: Some(serde_json::json!({"kind": "unit"})),
421            idempotency_key: idempotency_key.map(str::to_owned),
422        }
423    }
424
425    fn compact_task_has_expected_parts(
426        result: &Result<PgTask<CompactType>, FromRowError>,
427    ) -> AssertionResult {
428        match result {
429            Ok(task)
430                if task.args == b"payload"
431                    && task.parts.attempt.current() == 2
432                    && task.parts.status.load() == Status::Pending
433                    && task.parts.ctx.priority() == 7
434                    && task.parts.ctx.lock_by() == &Some("worker-a".to_owned())
435                    && task.parts.idempotency_key == Some("same-key".to_owned()) =>
436            {
437                Ok(())
438            }
439            Ok(task) => Err(AssertionError::new(vec![format!(
440                "unexpected task parts: {task:?}"
441            )])),
442            Err(error) => Err(AssertionError::new(vec![format!(
443                "expected successful conversion, got {error:?}"
444            )])),
445        }
446    }
447
448    fn compact_task_omits_idempotency_key(
449        result: &Result<PgTask<CompactType>, FromRowError>,
450    ) -> AssertionResult {
451        match result {
452            Ok(task) if task.parts.idempotency_key.is_none() => Ok(()),
453            Ok(task) => Err(AssertionError::new(vec![format!(
454                "expected idempotency_key None, got {:?}",
455                task.parts.idempotency_key
456            )])),
457            Err(error) => Err(AssertionError::new(vec![format!(
458                "expected successful conversion, got {error:?}"
459            )])),
460        }
461    }
462
463    fn column_not_found(column: &'static str) -> impl Fn(&FromRowError) -> AssertionResult {
464        move |error| match error {
465            FromRowError::ColumnNotFound(found) if found == column => Ok(()),
466            other => Err(AssertionError::new(vec![format!(
467                "expected missing column {column}, got {other:?}"
468            )])),
469        }
470    }
471
472    fn decode_error(error: &FromRowError) -> AssertionResult {
473        match error {
474            FromRowError::DecodeError(_) => Ok(()),
475            other => Err(AssertionError::new(vec![format!(
476                "expected decode error, got {other:?}"
477            )])),
478        }
479    }
480
481    fn unchecked_pool() -> PgPool {
482        let manager = ConnectionManager::<PgConnection>::new("postgres://127.0.0.1:1/not-used");
483        Pool::builder()
484            .max_size(1)
485            .connection_timeout(std::time::Duration::from_millis(10))
486            .build_unchecked(manager)
487    }
488
489    fn storage_uses_queue_and_buffer<Args, Codec, Fetcher>(
490        queue: &'static str,
491        buffer_size: usize,
492    ) -> impl Fn(&PostgresStorage<Args, Codec, Fetcher>) -> AssertionResult {
493        move |storage| {
494            if storage.config.queue().to_string() == queue
495                && storage.config.buffer_size() == buffer_size
496            {
497                Ok(())
498            } else {
499                Err(AssertionError::new(vec![format!(
500                    "expected queue {queue:?} and buffer {buffer_size}, got queue {:?} and buffer {}",
501                    storage.config.queue(),
502                    storage.config.buffer_size()
503                )]))
504            }
505        }
506    }
507
508    fn debug_mentions_public_type(result: &String) -> AssertionResult {
509        if result.contains("PostgresStorage")
510            && result.contains("config")
511            && !result.contains("pool")
512        {
513            Ok(())
514        } else {
515            Err(AssertionError::new(vec![format!(
516                "debug output did not describe storage without exposing the pool: {result}"
517            )]))
518        }
519    }
520
521    fn storage_for_type_name() -> PostgresStorage<String> {
522        let pool = unchecked_pool();
523        PostgresStorage::<String>::new(&pool)
524    }
525
526    fn storage_for_config(queue: &'static str, buffer_size: usize) -> PostgresStorage<String> {
527        let pool = unchecked_pool();
528        let config = Config::new(queue).set_buffer_size(buffer_size);
529        PostgresStorage::<String>::new_with_config(&pool, &config)
530    }
531
532    fn notify_storage_for_config(
533        queue: &'static str,
534        buffer_size: usize,
535    ) -> PostgresStorage<String, JsonCodec<CompactType>, PgNotify> {
536        let pool = unchecked_pool();
537        let config = Config::new(queue).set_buffer_size(buffer_size);
538        PostgresStorage::<String>::new_with_notify(&pool, &config)
539    }
540
541    fn cloned_storage_for_config(
542        queue: &'static str,
543        buffer_size: usize,
544    ) -> (PostgresStorage<String>, PostgresStorage<String>) {
545        let original = storage_for_config(queue, buffer_size);
546        let clone = original.clone();
547        (original, clone)
548    }
549
550    fn shares_lease_token(
551        pair: &(PostgresStorage<String>, PostgresStorage<String>),
552    ) -> AssertionResult {
553        let (original, clone) = pair;
554        if std::sync::Arc::ptr_eq(&original.lease_token, &clone.lease_token) {
555            Ok(())
556        } else {
557            Err(AssertionError::new(vec![
558                "clone did not share the per-process lease token".to_owned(),
559            ]))
560        }
561    }
562
563    fn cloned_storage(
564        pair: &(PostgresStorage<String>, PostgresStorage<String>),
565    ) -> AssertionResult {
566        storage_uses_queue_and_buffer("clone-api", 4)(&pair.1)
567    }
568
569    fn debug_storage() -> String {
570        format!("{:?}", storage_for_config("debug-api", 10))
571    }
572
573    fn storage_with_changed_codec() -> PostgresStorage<String, JsonCodec<CompactType>> {
574        storage_for_config("codec-api", 6)
575            .with_codec::<()>()
576            .with_codec::<JsonCodec<CompactType>>()
577    }
578
579    // Pins the codec type parameter to `()` at compile time. The `Codec = ()`
580    // slot is enforced by the parameter type; `Fetcher` stays inferred so the
581    // assertion does not depend on the (preserved) fetcher type's spelling.
582    fn pin_unit_codec<Fetcher>(_storage: &PostgresStorage<String, (), Fetcher>) {}
583
584    fn with_codec_swaps_to_unit_codec() -> String {
585        let pool = unchecked_pool();
586        let storage = PostgresStorage::<String>::new(&pool).with_codec::<()>();
587        // Compile-time check: this fails to compile if `with_codec::<()>` no
588        // longer yields a `()` codec slot, independent of `type_name` formatting.
589        pin_unit_codec(&storage);
590        storage.config.queue().to_string()
591    }
592
593    fn storage_accessors() -> (String, usize) {
594        let storage = storage_for_config("accessor-api", 8);
595        (
596            storage.config().queue().to_string(),
597            storage.config().buffer_size(),
598        )
599    }
600
601    fn basic_get_queue() -> String {
602        storage_for_config("basic-queue-api", 4)
603            .get_queue()
604            .to_string()
605    }
606
607    fn notify_get_queue() -> String {
608        notify_storage_for_config("notify-queue-api", 4)
609            .get_queue()
610            .to_string()
611    }
612
613    fn backend_trait_surfaces(notify: bool) -> (String, String, String) {
614        let worker = WorkerContext::new::<()>("backend-trait-worker");
615        if notify {
616            let storage = notify_storage_for_config("notify-trait-api", 2);
617            let middleware = std::any::type_name_of_val(&storage.middleware()).to_owned();
618            let heartbeat = std::any::type_name_of_val(&storage.heartbeat(&worker)).to_owned();
619            let stream = std::any::type_name_of_val(&storage.poll_compact(&worker)).to_owned();
620            (middleware, heartbeat, stream)
621        } else {
622            let storage = storage_for_config("basic-trait-api", 2);
623            let middleware = std::any::type_name_of_val(&storage.middleware()).to_owned();
624            let heartbeat = std::any::type_name_of_val(&storage.heartbeat(&worker)).to_owned();
625            let stream = std::any::type_name_of_val(&storage.poll_compact(&worker)).to_owned();
626            (middleware, heartbeat, stream)
627        }
628    }
629
630    fn exposes_accessors(result: &(String, usize)) -> AssertionResult {
631        if result.0 == "accessor-api" && result.1 == 8 {
632            Ok(())
633        } else {
634            Err(AssertionError::new(vec![format!(
635                "unexpected storage accessors: {result:?}"
636            )]))
637        }
638    }
639
640    fn constructs_backend_traits(result: &(String, String, String)) -> AssertionResult {
641        if result.0.contains("PgMiddleware")
642            && result.1.contains("Stream")
643            && result.2.contains("Stream")
644        {
645            Ok(())
646        } else {
647            Err(AssertionError::new(vec![format!(
648                "unexpected backend trait surfaces: {result:?}"
649            )]))
650        }
651    }
652
653    lets_expect! {
654        expect(crate_name()) {
655            to returns_the_crate_name { equal("apalis-diesel-postgres") }
656        }
657
658        expect(row(id, status, run_at, idempotency_key).try_into_task_compact::<Ulid, PgPool>()) {
659            let id = &Ulid::new().to_string();
660            let status = "Pending";
661            let run_at = Some(DateTime::now());
662            let idempotency_key = Some("same-key");
663
664            when row_has_all_required_fields {
665                to preserves_task_payload_and_context { compact_task_has_expected_parts }
666            }
667
668            when idempotency_key_is_absent {
669                let idempotency_key = None;
670                to omits_the_idempotency_key { compact_task_omits_idempotency_key }
671            }
672
673            when run_time_is_missing {
674                let run_at = None;
675                to rejects_the_row { be_err_and column_not_found("run_at") }
676            }
677
678            when status_is_unknown {
679                let status = "Unknown";
680                to rejects_the_row { be_err_and decode_error }
681            }
682
683            when id_is_not_a_ulid {
684                let id = "not-a-ulid";
685                to rejects_the_row { be_err_and decode_error }
686            }
687        }
688
689        expect(storage) {
690            let storage = storage_for_type_name();
691
692            when storage_is_built_from_the_task_type {
693                to uses_the_type_name_as_queue {
694                    storage_uses_queue_and_buffer(std::any::type_name::<String>(), 10)
695                }
696            }
697
698            when storage_is_built_with_an_explicit_config {
699                let storage = storage_for_config("public-api", 3);
700                to preserves_the_supplied_config { storage_uses_queue_and_buffer("public-api", 3) }
701            }
702
703            when storage_is_built_with_notify {
704                let storage = notify_storage_for_config("notify-api", 2);
705                to preserves_the_supplied_config { storage_uses_queue_and_buffer("notify-api", 2) }
706            }
707
708            when storage_is_cloned {
709                let storage = cloned_storage_for_config("clone-api", 4);
710                to keeps_the_queue_configuration { cloned_storage }
711                to shares_the_per_process_lease_token { shares_lease_token }
712            }
713        }
714
715        expect(debug_storage()) {
716            to describes_the_storage_without_exposing_the_pool { debug_mentions_public_type }
717        }
718
719        expect(storage_with_changed_codec()) {
720            to preserves_the_supplied_config { storage_uses_queue_and_buffer("codec-api", 6) }
721        }
722
723        // The codec swap to `()` is enforced at compile time by the
724        // `PostgresStorage<String, ()>` annotation in the helper; this leaf
725        // additionally confirms the swapped storage still constructs with the
726        // task-type-derived queue.
727        expect(with_codec_swaps_to_unit_codec()) {
728            to builds_a_unit_codec_storage_with_the_default_queue {
729                equal(std::any::type_name::<String>().to_owned())
730            }
731        }
732
733        expect(storage_accessors()) {
734            to exposes_the_queue_and_buffer_config { exposes_accessors }
735        }
736
737        expect(basic_get_queue()) {
738            to returns_the_basic_queue { equal("basic-queue-api".to_owned()) }
739        }
740
741        expect(notify_get_queue()) {
742            to returns_the_notify_queue { equal("notify-queue-api".to_owned()) }
743        }
744
745        expect(backend_trait_surfaces(notify)) {
746            let notify = false;
747
748            when basic_polling_storage {
749                to builds_heartbeat_middleware_and_compact_stream { constructs_backend_traits }
750            }
751
752            when notify_storage {
753                let notify = true;
754                to builds_heartbeat_middleware_and_compact_stream { constructs_backend_traits }
755            }
756        }
757    }
758}
apalis_diesel_postgres/lib.rs

apalis_diesel_postgres/
lib.rs
