Any way to get finsihed jobs?

[hldup]

Hey, I have an axum server that deploys jobs to different kinds of microservices. All it basically does is send jobs to the microservices and wait do other stuff in the meanwhile. Then once it has finished I want it to check up on the job I have dispatched to the microservice.
Is there a function or any way to query the RedisStorage to check if a job has finished?

[geofmureithi]

All storage have a `fetch_by_id` method. It should give you the updated version of a job.

…

On Mon, 1 Jul 2024 at 10:12, hldup ***@***.***> wrote: Hey, I have an axum server that deploys jobs to different kinds of microservices. All it basically does is send jobs to the microservices and wait do other stuff in the meanwhile. Then once it has finished I want it to check up on the job I have dispatched to the microservice. Is there a function or any way to query the RedisStorage to check if a job has finished? — Reply to this email directly, view it on GitHub <#340>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AWXVRGQO2BP4KCCZJDELLPDZKD6PFAVCNFSM6AAAAABKE6P6DGVHI2DSMVQWIX3LMV43ERDJONRXK43TNFXW4OZWHA4DINJWG4> . You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

[hldup]

Thanks for the reply! But i still have some problems

[hldup]

I have made a basic wrapper function that executes a trait function that is implemented to a struct that implements apalis::prelude::Job.
After the job has finished I still get the wrong data back from redis. What could I be doing wrong?

main.rs

     Monitor::<TokioExecutor>::new()
            .register_with_count(4, {
                WorkerBuilder::new("edit-runner")
                    .layer(TraceLayer::new())
                    .with_storage(storage.clone())
                    .data(state.clone())
                    .build_fn(jobhandler::run_job::<EditorJob, State>)
            })
            .run()
            .await
            .unwrap();

jobhandler::run_job()

pub async fn run_job<T, S>(mut job: T, state: Data<S>)
where
    T: Serialize + DeserializeOwned + Send + 'static + Unpin + Job + Sync + JobRunner<T, S>,
    S: StateHandler,
{
    match job.run(state).await {
        Ok(_) => {
            log::debug!("Job has finished ok");
            job.set_status(JobStatus::Ok)
        }
        Err(error) => {
            log::error!("job has failed");
            job.set_status(JobStatus::Error("failed".into()))
        }
    };
}

job.rs

impl JobRunner<EditorJob, crate::State> for EditorJob {
    async fn run(&mut self, state: Data<crate::State>) -> Result<(), jobhandler::JobError> {
        thread::sleep(Duration::from_secs(35));
        Ok(())
    }
    fn set_status(&mut self, status: jobhandler::JobStatus) {
        self.set_status(status);
    }
    fn get_status(&self) -> jobhandler::JobStatus {
        self.status().to_owned()
    }
}

[hldup]

Tho after it has run, it is recorded in the editor::job::done table.

[geofmureithi]

Does fetch_with_id + update work for you?
https://docs.rs/apalis-redis/0.6.0-alpha.0/src/apalis_redis/storage.rs.html#657-671

[hldup]

Is the job suppossed to automatically update after its done? @geofmureithi

[geofmureithi]

I am currently working on saving results. This issue #182 has some more context.

[hldup]

I migrated to the newest release as of now, I see that you implemented saving of results. (Great work, thank you!)
But when I fetch from the storage I get the initial state of the job not the result. Is there supposed to be a function that fetches results?
Received data from: jobhandler::node::Job:data
Instead of jobhandler::node::Job:data::result

[geofmureithi]

Its still WIP, I need to incorporate the result part. You could pull it manually if needed.

[hldup]

Thanks, I'll write my own method to fetch from redis till than.

[hldup]

I wrote a little code that fetches from the hashset and decodes the response but apalis returns a Encountered an error during streaming, any ideas what I could be doing wrong during decoding?

    let job: Value = match conn
        .hget(
            format!("{}:data::result", std::any::type_name::<T>()),
            id.clone().to_string(),
        )
        .await
    {
        Err(error) => {
            log::warn!(
                "Job for id: {} does not exist {}",
                id.to_string(),
                error
            );
            return Err(StatusCode::NOT_FOUND);
        }
        Ok(job) => job,
    };

    match job {
        Value::Nil => return Err(StatusCode::NOT_FOUND),
        Value::Data(job) => {
            match storage.get_codec().decode(&job) {
                Err(error) => {
                    log::error!("Failed to decode result from redis: {}", error);
                    return Err(StatusCode::INTERNAL_SERVER_ERROR);
                }
                Ok(job) => return Ok(job.job().to_owned().into()),
            };
        }
        _ => return Err(StatusCode::INTERNAL_SERVER_ERROR),
    };
    ```

[hldup]

Ah darn, I overlooked the data itself, the result is not saved as json. That's the main reason I get an error. stupid me

[kdesjard]

Is there an easy way to decode last_error back to Result<T,String> ?

[geofmureithi]

On Redis not yet, but on sql backends its stored as json for Result<String, String>, I need to improve this and am open to ideas.

[hldup]

AFAIK Only PostgreSQL has an explicit json format for columns so I think storing the result and error as a Json string is the best option

[kdesjard]

I tried the following and it worked for me:

In layers.rs:

impl<SV, A, T> Service<Request<T>> for AckService<SV, A, T>
...

if I convert the ok result to a serde_json::Value,instead of format!({ok:?}):
eg:

.map(|ok| serde_json::to_value(ok).unwrap_or_default())

and then in my own code, check last_error for a an option value, json decode the value back into a Result<MyType,String> and it works.

[hldup]

Yeah that's what I was thinking of too,
chaging this line in the code to not store the result as a string, instead just use serde_json to turn it into a value and then turn it into a string and save it in redis.

[kdesjard]

Ideally, the Error from the executed function would be put in the Result that is stored in the DB. I have my own Error type and what gets stored now is "FailedError: Unknown". Maybe I'm missing a trait or something.

[kdesjard]

Here is a changeset that is working for me (sqlite only right now) and currently has a couple of unwraps:

kdesjard@ecaac41

[geofmureithi]

@kdesjard Here is a PR that incorporates what you are trying to do.
#383