PostgreSQL adapter crashes worker after updating job status

[greyblake]

Hi!
First of all, thank you for the nice crate and amount of effort you put into it!

At the moment I am evaluating it for usage and building a prototype to see if it can serve our needs.
I run 1.0.0-beta with PostgreSQL adapter and ran into worker being crashed after a job is being processed successfully.

Starting worker...
START SendEmailJob: Hello World!!!!
FAILED SendEmailJob: Hello World!!!!


START SendEmailJob: Hello World!!!!
FINISHED SendEmailJob: Hello World!!!!

thread 'main' (2299142) panicked at /home/sergey/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/apalis-postgres-1.0.0-beta.1/src/ack.rs:151:58:
called `Result::unwrap()` on an `Err` value: RowNotFound
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

So the error originated in apalis-postgres/src/ack.rs#L151

Insights

apalis-postgres has 2 files that matches queries/task/lock_by_id.sql pattern:

• https://github.com/apalis-dev/apalis-postgres/blob/main/queries/task/lock_by_id.sql
• https://github.com/apalis-dev/apalis-postgres/blob/main/src/queries/task/lock_by_id.sql

The SQL in those files is different. I suspect SQLx is picking up a wrong file

[geofmureithi]

Thanks for reporting this. I am curious what your setup looks like because tests don't seem to catch this.
Are you using SharedPostgresStorage?

[greyblake]

@geofmureithi

✦ ❯ cat Cargo.toml  | rg apalis
name = "poc-apalis"
apalis = { version = "1.0.0-beta.1", features = ["catch-panic", "retry", "limit"] }
apalis-postgres = { version = "1.0.0-beta.1", features = ["tokio"] }

use std::time::Duration;

use apalis::layers::retry::HasherRng;
use apalis::layers::retry::backoff::MakeBackoff;
use apalis::layers::retry::{RetryPolicy, backoff::ExponentialBackoffMaker};
use apalis::prelude::*;
use apalis_postgres::PgPool;
use apalis_postgres::PostgresStorage;

use crate::config::Config;
use crate::jobs::{BuildReportJob, SendEmailJob};

const MAX_RETRIES: usize = 3;
const INITIAL_BACKOFF_MS: u64 = 3_000;
const MAX_BACKOFF_MS: u64 = 10_000;

/// Jitter factor is 10%
const JITTER: f64 = 0.1;

const DEFAULT_WORKER_CONCURRENCY: usize = 2;

pub async fn run_worker(config: Config) {
    println!("Starting worker...");

    // Connect to PostgreSQL
    let pool = PgPool::connect(&config.database_url)
        .await
        .expect("Failed to connect to database");

    // Create tables in schema `apalis` (e.g. `jobs`, `workers`, etc.)
    PostgresStorage::setup(&pool).await.unwrap();

    // Create storage backends for each job type
    let email_storage: PostgresStorage<SendEmailJob> = PostgresStorage::new(&pool);
    let report_storage: PostgresStorage<BuildReportJob> = PostgresStorage::new(&pool);

    let mut backoff_maker = ExponentialBackoffMaker::new(
        Duration::from_millis(INITIAL_BACKOFF_MS),
        Duration::from_millis(MAX_BACKOFF_MS),
        JITTER,
        HasherRng::default(),
    )
    .expect("Failed to create backoff maker");

    let email_backoff = backoff_maker.make_backoff();
    let report_backoff = backoff_maker.make_backoff();

    // Build and run the monitor with workers for both job types
    Monitor::new()
        .register(move |_| {
            WorkerBuilder::new("email-worker")
                .backend(email_storage.clone())
                .retry(RetryPolicy::retries(MAX_RETRIES).with_backoff(email_backoff.clone()))
                .concurrency(DEFAULT_WORKER_CONCURRENCY)
                .build(crate::jobs::handlers::handle_send_email_job)
        })
        .register(move |_| {
            WorkerBuilder::new("report-worker")
                .backend(report_storage.clone())
                .retry(RetryPolicy::retries(MAX_RETRIES).with_backoff(report_backoff.clone()))
                .concurrency(DEFAULT_WORKER_CONCURRENCY)
                .build(crate::jobs::handlers::handle_build_report_job)
        })
        .run_with_signal(tokio::signal::ctrl_c())
        .await
        .expect("Failed to run worker monitor");
}

[greyblake]

In addition to this the jobs have Done status in DB eventually:

poc_apalis=# SELECT * FROM apalis.jobs;
-[ RECORD 1 ]+-------------------------------------------------------------------------------------------------------------------------------------------------------------------
job          | \x7b226d657373616765223a2248656c6c6f20576f726c6421212121222c226d657461223a7b226661696c7572655f70726f626162696c697479223a302e352c226475726174696f6e223a313030307d7d
id           | 01KATV1TDK6V8EFXF4JYRG3D43
job_type     | poc_apalis::jobs::SendEmailJob
status       | Done
attempts     | 2
max_attempts | 5
run_at       | 2025-11-24 11:47:41+00
last_result  | {"Ok": null}
lock_at      | 2025-11-24 11:47:57.353238+00
lock_by      | email-worker
done_at      | 2025-11-24 11:47:58.356027+00
priority     | 0
metadata     | {}
-[ RECORD 2 ]+-------------------------------------------------------------------------------------------------------------------------------------------------------------------
job          | \x7b226d657373616765223a2248656c6c6f20576f726c6421212121222c226d657461223a7b226661696c7572655f70726f626162696c697479223a302e352c226475726174696f6e223a313030307d7d
id           | 01KATV4EGKGKYBSV98AKXHN99V
job_type     | poc_apalis::jobs::SendEmailJob
status       | Done
attempts     | 2
max_attempts | 5
run_at       | 2025-11-24 11:49:07+00
last_result  | {"Ok": null}
lock_at      | 2025-11-24 11:51:30.047096+00
lock_by      | email-worker
done_at      | 2025-11-24 11:51:31.049482+00
priority     | 0
metadata     | {}

[greyblake]

From what I've found so far.
The following SQL is executed:

UPDATE apalis.jobs
SET 
    status = 'Running',
    lock_at = now(),
    lock_by = $2
WHERE 
    status = 'Queued'
    AND run_at < now()
    AND id = ANY($1)
RETURNING *;

And sometimes (not always!) when it's executed, the job is already in a terminal state: either Done or Killed.
This causes the SQL update fail.

The issue can be reproduced even with a single worker with concurrency = 1.

[greyblake]

Ah, forgot to mention, that I haven't seen this error when I was running 0.7.4 version.

[geofmureithi]

@greyblake This should be resolved with #20

[greyblake]

@geofmureithi Thanks for the quick reply.
I've tested my protoype against #20, I can confirm that workers do not crush anymore.