dynamic job name

[ImSingee]

I want to use apalis-dashboard, and try to implement an API server.

However, I find that the storage is binding to a specific Job type which contains a const name. But I want to list all job names and construct the Job and Storage dynamically (rather than hard-code all job names inside API code).

Is there any plan to change the Job trait? For example, maybe change const Name to fn name()?

[geofmureithi]

Where are you experiencing the pain points?
At the API level, you can work with a custom trait.

apalis/examples/rest-api/src/main.rs

Lines 136 to 148 in d8ed4c9

	trait StorageRest<J>: Storage<Output = J> {
	fn name(&self) -> String;
	}
	impl<J, S> StorageRest<J> for S
	where
	S: Storage<Output = J> + JobStreamExt<J> + 'static,
	J: Job + Serialize + DeserializeOwned + 'static,
	{
	fn name(&self) -> String {
	J::NAME.to_string()
	}
	}

Is there any plan to change the Job trait? For example, maybe change const Name to fn name()?

Unless you mean fn name(&self) which am not sure about because would mean a jobs name is not known until loaded, so how did the storage and worker know the type?

You might also find using an enum helpful eg:

enum DynamicJob {
     Job1(Job1),
     Job2(Job2)
}

let job1 = service_fn(|req: JobRequest<Job1>| async move {
    Ok::<_, JobError>(())
});

let job1 = BoxService::new(job1);

let job2 = service_fn(|req: JobRequest<Job2>| async move {
    Ok::<_, JobError>(())
});
let job2 = BoxService::new(job2);

let mut svc = Steer::new(
    // All services we steer to
    vec![job1, job2],
    // How we pick which service to send the request to
    |req: &JobRequest<DynamicJob>, _services: &[_]| {
        match req.inner() {
            Job1(...) => 0,
            Job2(...) => 1
        }
    },
);

Learn more about steer here, https://docs.rs/tower/latest/tower/steer/index.html

Let me know if that helps

[geofmureithi]

I think the use of the term Name is the problem. Maybe we should call it Namespace

[geofmureithi]

See an example here: https://github.com/IgnisDa/ryot/blob/22d3505bb3ade0be6ae3b7a48f6c4a13166c437d/apps/backend/src/background.rs#L92

[ImSingee]

@geofmureithi Sorry for my lately reply.

My problem is, I want to build a seperate dashboard service, which do not know the job name during compilation time. I want it to dynamically discover all job names (or namespaces).

But in the example, nearly all API requires a Job trait (which must use a const name/namespace).

async fn get_jobs<J, S>(storage: web::Data<S>, filter: web::Query<Filter>) -> HttpResponse
    where
        J: Job + Serialize + DeserializeOwned + 'static,
        S: Storage<Output=J> + JobStreamExt<J> + Send,
{
    let storage = &*storage.into_inner();
    let mut storage = storage.clone();
    let counts = storage.counts().await.unwrap();
    let jobs = storage.list_jobs(&filter.status, filter.page).await;

    match jobs {
        Ok(jobs) => HttpResponse::Ok().json(JobsResult { jobs, counts }),
        Err(e) => HttpResponse::InternalServerError().body(format!("{e}")),
    }
}

For example, in this code snippet, I want to call storage.list_jobs. And it impl for PostgresStorage under a Job contraint.

impl<J: 'static + Job + Serialize + DeserializeOwned + Send + Sync + Unpin> JobStreamExt<J>
        for PostgresStorage<J>
    {
        async fn list_jobs(
            &mut self,
            status: &JobState,
            page: i32,
        ) -> Result<Vec<JobRequest<J>>, JobError> {
           // ...
     }
}

Seems these API (trait JobStreamExt) are defined just for API under a expose feature flag. Maybe we can make it loose? Like:

trait DynamicJob {
    fn name() -> &str
}

impl<J: DynamicJob> JobStreamExt<J>
        for PostgresStorage<J> {
    // ...
}

[geofmureithi]

Please note that the rest api is just an example and not part of the lib/crates.io, and hence your requirements would require you to implement some custom logic.
The DynamicJob would not help you because the storage can only be bound with J: Job.
Rust allows you to write custom traits. I would do:

trait MyJobStreamExt<D: DynamicJob> {
...
}

impl<J: Job + DynamicJob> MyJobStreamExt<J>
        for PostgresStorage<J> {
    // ... Do things with Dynamic job 
}

[ImSingee]

After digging a lot, I found you are right - There's too many Job::NAME without an object.

However, I still think, as a dashboard service, it's useful to support discover all jobs automatically. The only important part is name so there definitely can make some improves onJobStreamExt.

I created a demo for this: ImSingee@0e770ab

I do not see other storage implements, but I guess all are similar.

[geofmureithi]

The problem is that these changes need to propagate to other storages for parity. I think this is an edge case. You could do the implementation you have shown in your own crate. You can borrow from the current implementation, and reimplement your specific case. The expose feature will possibly be gone in v0.5 possibly replaced by a specific api crate.

[geofmureithi]

I have been thinking about this while working on #234.
There is no way out of something static. The main problem is that when starting a worker, we need to know which jobs to pull (eg from db). In postgress, different job types share the same table.

[ImSingee]

@geofmureithi I think the static name is required for worker, but not for dashboard. In micro service scenario, it's very common to have many different service (worker) do their own jobs and have their own job types. But when using a separate dashboard (like a manage/monitor service), it's very useful to dynamically discover all job types from all kinds of services (workers).

[geofmureithi]

Sure got it. I think what you are looking for can be resolved with a SQL query (eg in postgres)

SELECT DISTINCT job_type from workers;

[ImSingee]

Sure, it can be done by this easily😂 What I really hope is that apalis will support this method officially.

[geofmureithi]

apalis-board will be converted to a crate and will offer this feature 💯

[geofmureithi]

The problem is apalis is agnostic about the source. It just gets a stream and starts consuming.

[geofmureithi]

@ImSingee You will be happy to know that this has been fixed in v0.6.