Currently integrations tests are running on CI only for MySQL backend.

Let's add integration tests for the following backends

• [x] Mysql by @autotaker (added by #6)
• [x] Redis by @geofmureithi (added by #27)
• [x] Sqlite by @geofmureithi (added by #13)
• [x] Postgres by @autotaker (added by #11)

Let's strive to have different pull requests for each storage. This will help in collaboration. Also before starting on any backend, please add a comment and assigned.

Right now I do it manually. It would be nice if it was automatic. Steps needed:

1. PR to master
2. Running CI to master, must pass.
3. PR merge to master
4. Create new release
5. Publish crates to crates.io

Part of #9

This PR adds Test Suite with Postgres action and implements 5 tests for Postgres backend.

Tests

• test_consume_last_pushed_job
• test_acknowledge_job
• test_kill_job
• test_heartbeat_renqueueorphaned_pulse_last_seen_6min
• test_heartbeat_renqueueorphaned_pulse_last_seen_4min

Other changes

Refactoring

I extracted keep_alive_at(self, worker_id, last_seen) from keep_alive(self, worker_id) to improve testability.

Changed SQL statement to upsert worker in keep_alive:

When keep_alive_at(self, worker_id, last_seen) is called for worker_id already registered, the last_seen column of the worker row is updated with NOW() (system clock on the Postgres server). Although this behavior was trivial things, the value of last_seen can be different from the current time due to the refactoring.

Therefore, I changed the updated value to EXCLUDED.last_seen, which is the same value as it is inserted if the worker_id is registered at first time.

 ON CONFLICT (id) DO
-  UPDATE SET last_seen = NOW()
+  UPDATE SET last_seen = EXCLUDED.last_seen

The storage api for mysql doesnt support this feature.

    // Worker not seen in 5 minutes yet has running jobs
    StorageWorkerPulse::RenqueueOrpharned { count: _ } => {
    //...
    Ok(true)

The goal is to rewrite the query at mysql.rs#L202 to be Mysql compatible and uncomment it.

I want to build multiple async function works. How can I do this? The following cannot be achieved

Monitor::new()
    .register_with_count(6, move |_| {
        WorkerBuilder::new(worker_storage.clone())
            .layer(SentryJobLayer)
            .layer(TraceLayer::new())
            .build_fn(queue::send_email)
            .build_fn(queue::send_message)
            .build_fn(queue::and_more)
    })

I cant find any apalis examples with using extensions for shared data.

Only a wrong example in docs:

/// Extension data for jobs.
///
/// forked from [axum::Extensions]
/// # In Context
///
/// This is commonly used to share state across jobs.
///
/// ```rust,ignore
/// use apalis::{
///     Extension,
///     WorkerBuilder,
///     JobContext
/// };
/// use std::sync::Arc;
///
/// // Some shared state used throughout our application
/// struct State {
///     // ...
/// }
///
/// async fn email_service(email: Email, ctx: JobContext) {
///     let state: &Arc<State> = ctx.data_opt().unwrap();
/// }
///
/// let state = Arc::new(State { /* ... */ });
///
/// let worker = WorkerBuilder::new(storage)
///     .layer(Extension(state))
///     .build_fn(email_service);
/// ```

Where even

use apalis::{
    Extension,
     WorkerBuilder,
     JobContext
 };

is throwing error apalis does not have Extension.

Please make a working example with Shared state.

Currently apalis-sql is using version 0.5 of sqlx this make using it a bit harder than necessary to use in an application using sqlx 0.6.

I'm unable to use the PostgresStorage::new method do to the differing sqlx versions.

Workaround

Pass a connection string to PostgresStorage::connect.

This PR fixes the issue sharing the connection pool on MySQL backend, and adds more tests for MySQL backend.

This PR also contains a bug fix that lock_at is not updated when a job is locked. (See stream_jobs method)

#19 This PR adds a workflow (Continuous delivery) to publish crates to crates.io.

Spec

Trigger

Continuous delivery workflow is triggered by pushing tags to the remote repository. Tag must be of the form vX.Y.Z or vX.Y.Z-SUFFIX. A tag with suffix means pre-release version and it enables --dry-run mode.

Secrets

Please set CARGO_REGISTRY_TOKEN for the authentication token for crates.io.

Jobs

• test
  • This job runs several tests defined in ci.yaml.
• publish
  • This job publishes crates (apalis-core, apalis-cron, apalis-redis, apalis-sql, and apalis) in an appropriate order.
  • If there are some inconsistencies between the tag version and a package version (written in Cargo.toml), the job fails before pushing the crates.

Test

I have created v0.3.4-rc10 tag and confirmed that the workflow succeeds. https://github.com/autotaker/apalis/actions/runs/3355470587

Other changes

• Trigger of ci.yaml is changed
  • CI for topic branch runs only after creating a pull request. (push trigger is limited to master/develop)
  • Add workflow_call trigger to call CI workflow from CD workflow.

This PR fixes a bug with killing a job.

Bug details

In MySQL/Postgres/Sqlite3 backends, Storage::kill(self, worker_id, job_id) sets the status to 'Kill', which is incompatible to the definition of JobStatus::Killed. https://github.com/geofmureithi/apalis/blob/3cfc8df9a9bcb288ea24c3892e2446c45247b947/packages/apalis-core/src/request.rs#L12-L26

As a result, deserialization of job context will panic while converting Kill to JobState. https://github.com/geofmureithi/apalis/blob/3cfc8df9a9bcb288ea24c3892e2446c45247b947/packages/apalis-sql/src/from_row.rs#L53

I found a bug with table name in mysql.rs.

In default configuration (lower_case_table_names=0) of MySQL on linux, table name are case-sensitive. The table name managing jobs should be jobs, but in some SQL it is referred as Jobs.

As a result, this library does not work with such MySQL servers.

With this PR, I modified every occurrence of Jobs in SQL statements with jobs.

Hi! I found your strange example without any code. And nobody asks about it.

May be I miss something and now everybody should use telepathy to use examples? :)

twitter-screenshot-bot

It is currently not possible to use apalis on the same postgres database as the main application.

This is do to the sqlx Migrator currently not support multiple applications using the same database. As the postgres migrations are already using a separate schema, it would be quite nice to be able to setup apalis in the same database as the main application it self.

I currently see two options:

1. Users have to copy our migrations into there sqlx migration folder.
2. We don't use the sqlx migrator for postgres until sqlx supports mutliple applications in the same database.

This isn't a problem for sqlite or mysql as either of these support postgresql like schemas.

See: https://github.com/launchbadge/sqlx/issues/1698

Would be great if surrealdb support can be added. It can run in memory, single node with file as well as distributed, has events support and is written in Rust.

SurrealDB in 100 Seconds gives a good overview.- https://www.youtube.com/watch?v=C7WFwgDRStM

When trying to terminate the application with Ctrl + C (or by sending SIGINT directly) the monitor shuts down, but the application is stuck until it receives a SIGTERM signal.

$ cargo run --package axum-example
    Finished dev [unoptimized + debuginfo] target(s) in 0.18s
     Running `target/debug/axum-example`
2022-10-07T13:33:36.526309Z DEBUG axum_example: listening on 127.0.0.1:3000
2022-10-07T13:33:36.526681Z DEBUG apalis_core::worker::monitor: Listening shut down command (ctrl + c)
2022-10-07T13:33:46.759826Z DEBUG apalis_core::worker::monitor: Workers shutdown complete
^C^C

I tested this with the axum example and my own app, both have the same result.

The actix-web example does properly exit, so maybe axum needs some special handling for signals.

$ cargo run --package actix-web-example
    Finished dev [unoptimized + debuginfo] target(s) in 0.19s
     Running `target/debug/actix-web-example`
[2022-10-07T13:40:11Z INFO  actix_server::builder] Starting 4 workers
[2022-10-07T13:40:11Z INFO  actix_server::server] Actix runtime found; starting in Actix runtime
[2022-10-07T13:40:11Z DEBUG apalis_core::worker::monitor] Listening shut down command (ctrl + c)
[2022-10-07T13:40:13Z INFO  actix_server::server] SIGINT received; starting forced shutdown
[2022-10-07T13:40:13Z INFO  actix_server::worker] Shutting down idle worker
[2022-10-07T13:40:13Z DEBUG actix_server::accept] Paused accepting connections on 127.0.0.1:8000
[2022-10-07T13:40:13Z INFO  actix_server::worker] Shutting down idle worker
[2022-10-07T13:40:13Z INFO  actix_server::accept] Accept thread stopped
[2022-10-07T13:40:13Z INFO  actix_server::worker] Shutting down idle worker
[2022-10-07T13:40:13Z INFO  actix_server::worker] Shutting down idle worker
[2022-10-07T13:40:13Z DEBUG apalis_core::worker::monitor] Workers shutdown complete

Currently it only shows one worker send_email. Would be good to showcase more types of jobs being consumed.

    Monitor::new()
        .register_with_count(5, move |_| {
            WorkerBuilder::new(sqlite.clone())
                .layer(TraceLayer::new())
                .build_fn(send_email)
        })
        .run()
        .await