[Bug]: Integration tests flaky: apalis jobs stolen by wrong workers due to shared job_type dispatch

[ALAWIII]

What happened?

Description:

Problem

When running integration tests in parallel, each test creates unique workers (via Uuid::new_v4()) and waits for its jobs to finish by polling lock_by IN (worker_a, worker_b). However, apalis dispatches jobs based on job_type, not worker_id. This means any worker registered with worker_type = "family_cloud::jobs::DeleteJob" can steal jobs from another test's workers.

Root Cause

From inspecting apalis.jobs and apalis.workers during a parallel test run:

• 8 workers were registered (4× DeleteJob, 4× CopyJob) across 6 parallel tests
• All 11 jobs were processed by only 2 worker UUIDs
• The remaining 6 worker UUIDs never picked up any jobs
• Tests waiting on those 6 UUIDs via lock_by looped until timeout

Observed Behavior

• 2 out of 6 integration tests always fail with "worker did not finish in time"
• Increasing timeout (up to 600s) has no effect — the jobs were already processed by a different worker

Expected behavior

Each test should be able to reliably wait for its own enqueued jobs to finish.

Steps to reproduce

1. Create multiple integration tests that each call init_workers() (generates unique worker UUIDs per test)
2. Each test enqueues jobs via an API endpoint that dispatches DeleteJob / CopyJob
3. Each test waits for its jobs by polling apalis.jobs WHERE lock_by IN (worker_a_uuid, worker_b_uuid)
4. Run all tests in parallel: cargo test

Expected

Each test's jobs are processed by its own workers and lock_by matches the test's worker UUIDs.

Actual

Apalis dispatches jobs by job_type — any worker with matching worker_type picks up the job regardless of UUID. 2 out of 6 tests always timeout because their enqueued jobs were processed by another test's workers, so lock_by never matches their UUIDs.

Reproducible: Yes, consistently 2/6 tests fail on every parallel run.

Minimal code example

pub async fn init_workers(con: &PgPool, rfs: Client) -> anyhow::Result<WorkersName> {
    let n_worker = WorkersName {
        delete: Uuid::new_v4().to_string(),
        copy: Uuid::new_v4().to_string(),
    };
    init_apalis(con, rfs, n_worker.clone()).await?;
    Ok(n_worker)
}
async fn wait_job_until_finishes(con: &PgPool, workers: &WorkersName) -> anyhow::Result<()> {
    tokio::time::timeout(Duration::from_secs(25), async {
        loop {
            let value = sqlx::query!(
                r#"SELECT (
                    -- Phase 1: jobs must exist first
                    (SELECT COUNT(*) FROM apalis.jobs j
                     WHERE j.lock_by IN ($1,$2)) > 0
                    AND
                    -- Phase 2: all of them must be done
                    (SELECT COUNT(*) FROM apalis.jobs j
                     WHERE j.lock_by IN ($1,$2) AND j.done_at IS NULL) = 0
                ) AS "done!: bool""#,
                workers.delete,
                workers.copy,
            )
            .fetch_one(con)
            .await?;
            if value.done {
                break;
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
        Ok::<_, sqlx::Error>(())
    })
    .await
    .expect("worker did not finish in time")?;
    Ok(())
}

Version

1.0.0-rc.x

Environment

• OS: Linux rusty 6.19.2-2-cachyos V0.3 apalis#1 SMP PREEMPT_DYNAMIC Mon, 16 Feb 2026 20:41:55 +0000 x86_64 GNU/Linux
• Rust version: rustc 1.90.0 (1159e78c4 2025-09-14)
• Cargo version: cargo 1.90.0 (840b83a10 2025-07-30)

Relevant log output

Possible Solutions

• Wait by job_type + run_at >= enqueued_at timestamp instead of lock_by
• Capture and track job IDs at enqueue time
• Use a unique apalis schema per test to fully isolate job queues

log tables:
jobs_table.xlsx
workers_table.xlsx

[ALAWIII]

I am wondering if it's possible to add configuration settings to apalis structs like:
.lock_job_by_worker_id(true) or something similar.

let delete_worker = WorkerBuilder::new(w_names.delete)
        .backend(delete_backend.clone())
        .lock_job_by_worker_id(true)
        .retry(RetryPolicy::retries(usize::MAX).with_backoff(backoff_maker.make_backoff()))
        .concurrency(cpus * 4)
        .data(jobstate.clone())
        .build(delete_file_rustfs);

[geofmureithi]

The recommendation is to use the command-line flag --test-threads or the environment variable RUST_TEST_THREADS when running tests. You could also create different job_types per test.

…

On Thu, 26 Feb 2026 at 15:05, ALAWIII ***@***.***> wrote: *ALAWIII* left a comment (apalis-dev/apalis-postgres#60) <#60 (comment)> I am wondering if it's possible to add configuration settings to apalis structs like: .lock_job_by_worker_id(true) or something similar. let delete_worker = WorkerBuilder::new(w_names.delete) .backend(delete_backend.clone()) .lock_job_by_worker_id(true) .retry(RetryPolicy::retries(usize::MAX).with_backoff(backoff_maker.make_backoff())) .concurrency(cpus * 4) .data(jobstate.clone()) .build(delete_file_rustfs); — Reply to this email directly, view it on GitHub <#60 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AWXVRGVYV53V6BWODFNYEID4N3ORLAVCNFSM6AAAAACWAOK2GKVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZTSNRWGE4TONRWGI> . You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

[ALAWIII]

create different job_types per test

I used to use DeleteJob struct as job_type annotation for the worker builder !!
can we rename job_types ? if yes, please teach me how.

[geofmureithi]

You can set a specific namespace https://github.com/apalis-dev/apalis-postgres/blob/484fbff9fc33ef2e40ec97448ce9b4bf4cac4594/examples/pubsub.rs#L17

…

On Thu, 26 Feb 2026 at 15:20, ALAWIII ***@***.***> wrote: *ALAWIII* left a comment (apalis-dev/apalis-postgres#60) <#60 (comment)> create different job_types per test I used to use DeleteJob struct as job_type annotation for the worker builder !! can we rename job_types ? if yes, please teach me how. — Reply to this email directly, view it on GitHub <#60 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AWXVRGQHCHWYIAD3IEHVJ2T4N3QHHAVCNFSM6AAAAACWAOK2GKVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZTSNRWGI3DINZUGM> . You are receiving this because you commented.Message ID: ***@***.***>

[ALAWIII]

adding .make_shared_with_config(Config::new(&w_names.delete)) sets the worker_type column in apalis.worker table and job_type column in apalis.jobs table .
so that worker_type = job_type columns , and in my case I set it also to be the worker name !

    debug!("creating shared storage for multiple workers.");
    let mut store = SharedPostgresStorage::new(con.clone());
    let e_msg = "failed to create a shared backend: ";
    debug!("creating delete/copy backends for the two distinct jobs.");
    let delete_backend = store
        .make_shared_with_config(Config::new(&w_names.delete))
        .map_err(|e| JobError::SharedCreation(anyhow::anyhow!("{}{}", e_msg, e)))?;

 let delete_worker = WorkerBuilder::new(w_names.delete)
        .backend(delete_backend.clone())
        .retry(RetryPolicy::retries(usize::MAX).with_backoff(backoff_maker.make_backoff()))
        .concurrency(cpus * 4)
        .data(jobstate.clone())
        .build(delete_file_rustfs);

And now I query finished jobs based on job_type directly !
Bonus lock_by column will always equal job_type as desired.

SELECT
	(
		-- Phase 1: jobs must exist first
                    (
			SELECT
				COUNT(*)
			FROM
				apalis.jobs j
			WHERE
				j.job_type IN ($1, $2) 
                     ) > 0
		AND
		-- Phase 2: all of them must be done
                    (
			SELECT
				COUNT(*)
			FROM
				apalis.jobs j
			WHERE
				j.job_type IN ($1, $2) AND j.done_at IS NULL 
                     ) = 0
	) AS "done!: bool"

The problem was solved successfully .