[Bug]: Intermittent issue causing failure to decode jobs

[m4c-ro]

What happened?

I'm having an issue where apalis doesn't run any background jobs from a particular worker, but the missing jobs reappear after a restart. This issue is occurring approximately every third time I run my program. I'm using the sqlite backend.

Looking into it I started calling list_tasks. In the error case, the very first job I enqueue immediately causes the call to list_tasks to fail with a decode error:

Decode(DecodeError(DecodeError(Syntax("invalid type: sequence, expected unit"))))

I think this is coming from serde_json bubbling up from apalis-sql/src/from_row.rs. Jobs are processed after I restart, so maybe there is some state outside of the database which is corrupted. But here is the state of the jobs table when this error occurred anyway:

 job       | id                         | job_type | status | attempts | max_attempts | run_at     | last_result | lock_at    | lock_by | done_at    | priority | metadata
-----------+----------------------------+----------+--------+----------+--------------+------------+-------------+------------+---------+------------+----------+----------
<--snip--> | 01KQYJKD4B4Z0QMHWM51XQZSSM | email    | Done   | 1        | 5            | 1778068862 | {"Ok":null} | 1778068862 | email   | 1778068862 | 0        | {}

I enabled tracing at the TRACE level, and added the tracing layer to the worker, but unfortunately I'm not seeing any logs related to this issue.

Crate versions:

apalis = { version = "1.0.0-rc.7", default-features = false, features = ["catch-panic", "limit", "retry", "timeout", "tracing"] }
apalis-codec = { version = "0.1.0-rc.7", default-features = false, features = ["msgpack"] }
apalis-core = "1.0.0-rc.7"
apalis-cron = "1.0.0-rc.7"
apalis-sqlite = "1.0.0-rc.7"

Any ideas?

Expected behavior

Jobs to be run reliably

Steps to reproduce

Not sure yet. This seems to happen randomly.

Minimal code example

Version

1.0.0-rc.x

Environment

• OS: Linux
• Rust version:
• Cargo version:

Relevant log output

Additional context

No response

[m4c-ro]

I'm was using MsgPackCodec, but trying with the JSON one now I actually get a different error from list_tasks:

Decode(DecodeError(Error("invalid type: map, expected unit", line: 1, column: 0))

Here is a row from the database using JSON, including the full job. The JSON in the job column looks good to me.

 job                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               | id                         | job_type | status | attempts | max_attempts | run_at     | last_result                    | lock_at    | lock_by | done_at    | priority | metadata
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+----------------------------+----------+--------+----------+--------------+------------+--------------------------------+------------+---------+------------+----------+----------
 {"email":{"from":"noreply@test.local","reply_to":"support@test.local","to":"OBdRbSLtdgnnZLqiOiLmHw==@test.local","subject":"Email Verification","content_html":"<a href=\"https://test.local/email/verify?t=9e9igs2dNLPv6Gh8vUZTWJWbwTYClg637HDEDu3mQ20ED0TPr_DGmkrigaZVfvRhQBIDGtJLeDnkA7qp4LFqJg==\">Verify Email</a>","content_text":"Verify Email: https://test.local/email/verify?t=9e9igs2dNLPv6Gh8vUZTWJWbwTYClg637HDEDu3mQ20ED0TPr_DGmkrigaZVfvRhQBIDGtJLeDnkA7qp4LFqJg==","smtp_hostname":"smtp4dev","smtp_relay":true}} | 01KQYQPC7EZD3QRFTM6C8N8WMF | email    | Done   | 1        | 5            | 1778074202 | {"Ok":null}                    | 1778074202 | email   | 1778074202 | 0        | {}

[geofmureithi]

Several things:

1. How does your task struct look like?
2. How are you pushing your tasks.

If possible could you include a minimal code that I can use to reproduce (eg how you are connecting apalis-cron too)

[geofmureithi]

The error: invalid type: map, expected unit shows a serialization issue. I would want to know what your task handler looks like too.

[m4c-ro]

Here is the task struct (EmailJob):

#[derive(Serialize, Deserialize, Clone)]
pub struct EmailJob {
    pub email: Email,
}

#[derive(Serialize, Deserialize, Clone)]
pub struct Email {
    from: String,
    reply_to: String,
    to: String,
    subject: String,
    content_html: String,
    content_text: String,
    smtp_hostname: String,
    smtp_relay: bool,
}

I'm pushing the tasks using the push method from TaskSink on:

SqliteStorage<EmailJob, JsonCodec<Vec<u8>>, SharedFetcher<Vec<u8>>>
// OR
SqliteStorage<EmailJob, MsgPackCodec, SharedFetcher<Vec<u8>>>

I am using SharedSqliteStorage to create instances of SqliteStorage.

I'll put together code for repro.

[m4c-ro]

I've managed to reproduce this pretty simply. It takes quite a few runs to find the error, but it does happen.

Example Output

Delayed for 1 seconds, this seems to be required
Couldn't list tasks: Decode(
    DecodeError(
        Error("invalid type: map, expected unit", line: 1, column: 0),
    ),
)

use std::time::Duration;

use apalis::prelude::*;
use apalis_codec::json::JsonCodec;
use apalis_core::backend::shared::MakeShared;
use apalis_core::task_fn::task_fn;
use apalis_sqlite::shared::SharedFetcher;
use apalis_sqlite::{Config, SharedSqliteStorage, SqliteStorage};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Clone)]
struct EmailJob {
    field: String,
}

impl EmailJob {
    async fn run(self) {}
}

#[tokio::main]
async fn main() {
    // I'm actually using a file, but this still works for reproduction
    let mut storage = SharedSqliteStorage::new(":memory:");
    SqliteStorage::setup(storage.pool()).await.unwrap();

    let mut backend: SqliteStorage<EmailJob, JsonCodec<Vec<u8>>, SharedFetcher<Vec<u8>>> =
        storage.make_shared_with_config(Config::new("email")).unwrap();

    let service = task_fn(EmailJob::run);

    let worker = WorkerBuilder::new("email")
        .backend(backend.clone())
        .build(service);

    tokio::spawn(async move {
        let _ = worker.run().await;
    });

    let result = storage.make_shared_with_config(Config::new("debug"));
    let monitor_backend: SqliteStorage<(), JsonCodec<Vec<u8>>, SharedFetcher<Vec<u8>>> =
        result.unwrap();
    tokio::spawn(async move {
        loop {
            tokio::time::sleep(Duration::from_millis(50)).await;
            
            let filter = apalis_core::backend::Filter {
                status: None,
                page: 1,
                page_size: Some(10),
            };
            let result = monitor_backend.list_tasks("email", &filter).await;
            if let Err(error) = result {
                println!("Couldn't list tasks: {error:#?}");
                std::process::exit(0);
            } else {
                println!("No error yet... Try running again (the issue is intermittent)");
                std::process::exit(0);
            };
        }
    });

    tokio::time::sleep(Duration::from_millis(50)).await;

    let email_job = EmailJob {
        field: String::new(),
    };
    backend.push(email_job).await.unwrap();
}

Edit: Simplified further

[geofmureithi]

Are you sure this is not the cause?

let monitor_backend: SqliteStorage<(), JsonCodec<Vec<u8>>, SharedFetcher<Vec<u8>>>

The types need to match

[m4c-ro]

Aha. Yes, that does fix the error. But unfortunately that means the underlying issue is somewhere else.

I'm still having the issue with background jobs not running intermittently. I'll keep trying to figure this out.

[geofmureithi]

Ok noted, let me give you a patch to see if it helps you.

[geofmureithi]

Can you give a try to the patch in #76?

[m4c-ro]

I think I've found the culprit. The worker for the job was actually exiting due to a "WORKER_ALREADY_EXISTS" error from src/queries/register_worker.rs. Therefore, "missing jobs". I'm going to close the issue since it's clearly a bug on my end, thank you very much @geofmureithi for the help.

In case anyone finds this useful:

• If a matching worker already exists in the database on app startup, apalis will not start the worker task
• This can be caused by apalis not removing the row from the database during a previous run
• Understandably this would happen during a crash, but for some reason even during normal app shutdown the row sometimes get lefts behind in my case. This is probably due to some bad cleanup code on my end.
• Unfortunately for me worker exit isn't logged by apalis, so I wasn't seeing this happen until I checked the result from each workers task
• Adding a variable suffix to worker names (EG: worker-xyz-instance-123-run-456) might help prevent this issue during crash recovery, although I'm not sure if any other uniqueness constraints are checked here

Thanks again @geofmureithi.