Store Job result

[kdesjard]

How would I go about storing the result of a job in the DB along with the request?

Thanks.

[geofmureithi]

Usually a layer should work well.
You can wrap a layer and call update_by_id inside it
See also
#142 (comment)

[kdesjard]

I can get the storage context if I stuff it into a layer but how would I get the JobRequest from inside the build_fn function?

Looking at the update_by_id code in each db implementation, I don't see it updating the "job" column. Ideally the results would be a separate column.

[geofmureithi]

It is intentional that you can't. This is to keep the job_fn sort of pure. If you had a JobRequest and mutated it, nothing would happen and this may give the illusion that the request was updated and I wanted to avoid that.

[geofmureithi]

The API is still a work in progress. This will be improved in the upcoming 0.5 release.

[geofmureithi]

Here is how to do it with Layer:

pub struct SaveResultLayer;

impl<S> Layer<S> for SaveResultLayer {
    type Service = SaveResultService<S>;

    fn layer(&self, service: S) -> Self::Service {
        SaveResultService {
            service,
        }
    }
}

pub struct SaveResultService<S> {
   service: S,
}

impl<S, J> Service<JobRequest<J>> for SaveResultService<S>
where
    S: Service<JobRequest<J>>,
    Request: std::fmt::Debug,
    J: Job
{
   type Response = S::Response;
   type Error = S::Error;
   type Future = S::Future; // Possibly box dyn here

   fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
       self.service.poll_ready(cx)
   }

   fn call(&mut self, request: JobRequest<J>) -> Self::Future {
       let id = request.id();
       let storage: PostgresStorage<J> = request.data();
       let fut = self.service.call(request);
       Box::pin(async move {
        let res = fut.await;
        let job = storage.get_by_id(&id).await;
        job.set_result(&res.to_string());
        storage.update_by_id(&id, job).await;
        res
       })
   }
}

I have not tested it but, it should give you the idea.

[kdesjard]

I see there is a new update trait/function for Storage. I have stored the SqliteStorage into the worker's Data, and then I'm trying to use this in the executor to update the job results:

let db = &data.db;                                                                                                                                                                                                                    
let job = Request::new(job.clone());                                                    
db.update(job).await;

But then I get "Missing SqlContext". Is there a simple way now to update a job's result after execution?

Thx

[geofmureithi]

You need to use new_with_data. You can get SqlContext from the previous request as described in #271

[kdesjard]

I got it to work with "new_with_data", however, it's not very ergonomic way of doing it. Also, I'm not able to update the job payload as the update methods don't update the "job" column, only status, attempts etc:

"UPDATE jobs SET status = ?, attempts = ?, done_at = ?, lock_by = ?, lock_at = ?, last_error = ? WHERE id = ?"

It is really critical to be able to update the job to store results.

[geofmureithi]

This has been resolved in the v0.6.0 release

[kdesjard]

Could you please describe how this is done? I don't see anywhere in 0.6.0-X where the job column is updated. In each storage type the update method does not update the job column:

eg: sqlite:

        let query =
                "UPDATE Jobs SET status = ?1, attempts = ?2, done_at = ?3, lock_by = ?4, lock_at = ?5, last_error = ?6 WHERE id = ?7";
        sqlx::query(query)
...

I would expect to see (which is what I've done in my own local patches and may not be the best way):

        let job = self
            .codec
            .encode(&job.inner())
            .map_err(|e| sqlx::Error::Io(io::Error::new(io::ErrorKind::InvalidData, e)))?;
...

        let query =
                "UPDATE Jobs SET job = ?1, status = ?2, attempts = ?3, done_at = ?4, lock_by = ?5, lock_at = ?6, last_error = ?7 WHERE id = ?8";
        sqlx::query(query)
            .bind(job)
...

[geofmureithi]

In sqlite this is how it works:

apalis/packages/apalis-sql/src/sqlite.rs

Line 481 in 95618ff

impl<T: Sync + Send> Ack<T> for SqliteStorage<T> {

The only thing I need is to update the name of the column name from last_error to last_result.

[kdesjard]

Ok, I see what you're doing now. Thanks!