Question about persisted crons

[twuebi]

Hi and thanks for the awesome libary!

At lakekeeper we're currently working on integrating some form of persisted cron-scheduling both for user schedulable tasks as well as some fixed-schedule background tasks. For both kinds of tasks, it is important to attach some data to the cron schedule, i.e. the receiver functions needs to have some data to do their work.

Is this something that's possible today with this library?

From what I understand after skimming some of the apalis sources and having taken a look at the cron example, we'd have to spawn a worker per schedule? I also understand that the cron definitions themselves are not persisted but only its current incancation? The only associated data with cron ticks is the datetime of the tick?

Kind regards and thank you
Tobias

[geofmureithi]

Thanks for the kind words and good to know you are considering using this lib.
First, I am curious if you have looked at the persisted cron example.

For both kinds of tasks, it is important to attach some data to the cron schedule, i.e. the receiver functions needs to have some data to do their work.

This should be possible with the persistence, as data saved is generic. You need to create a type to hold all the metadata you want. If you can provide more info on the metadata, we can discuss the best approaches.
There are some changes upcoming in the next version related to this on this PR

we'd have to spawn a worker per schedule?

Currently yeah. But apalis is pretty extensible, it should be straight forward to implement a custom Backend to represent multiple shedules per worker scenario. I am open to helping you achieve this.

[twuebi]

Thanks for the fast reply!

First, I am curious if you have looked at the persisted cron example.

I did, I've tried to alter it according to my use-case, I've attached some code here:

#[cfg(test)]
mod test {
    use apalis::prelude::{Data, Monitor, WorkerBuilder, WorkerFactoryFn};
    use apalis_cron::{CronStream, Schedule};
    use apalis_sql::postgres::PostgresStorage;
    use serde::{Deserialize, Serialize};
    use sqlx::PgPool;
    use std::str::FromStr;
    use uuid::Uuid;

    #[derive(Debug, Serialize, Deserialize)]
    pub struct StatsTask {
        warehouse_ident: Uuid,
    }

    pub async fn collect_stats(StatsTask { warehouse_ident }: StatsTask, pg_pool: Data<PgPool>) {}

    #[sqlx::test]
    pub async fn test(pg_pool: PgPool) {
        let mut mon = Monitor::new();
        let cr = Schedule::from_str("1 * * * * *").unwrap();
        let cron = CronStream::new(cr);
        PostgresStorage::setup(&pg_pool).await.unwrap();
        let store = PostgresStorage::new(pg_pool.clone());
        let backend = cron.pipe_to_storage(store);
        let be = mon
            .register(
                WorkerBuilder::new("stats")
                    .data(pg_pool)
                    .backend(backend)
                    .build_fn(collect_stats),
            )
            .run()
            .await
            .unwrap();
    }
}

This currently doesn't compile due to:

54  |     let backend = cron.pipe_to_storage(pool);
    |                        ^^^^^^^^^^^^^^^ the trait `From<chrono::DateTime<chrono::Utc>>` is not implemented for `service::stats::StatsTask`

which is in-line with what I saw when skimming the apalis-cron code, currently the cron data source only provides the timestamp of the tick to the consuming function.

[twuebi]

Ah, I think I get now how that'd work, since schedule : worker is 1 : 1, I'd just supply my task data as a data, so something like:

#[cfg(test)]
mod test {
    use apalis::prelude::{Data, Monitor, WorkerBuilder, WorkerFactoryFn};
    use apalis_cron::{CronStream, Schedule};
    use apalis_sql::postgres::PostgresStorage;
    use chrono::Utc;
    use serde::{Deserialize, Serialize};
    use sqlx::PgPool;
    use std::str::FromStr;
    use uuid::Uuid;

    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct StatsTask {
        warehouse_ident: Uuid,
    }

    pub async fn collect_stats(
        now: chrono::DateTime<Utc>,
        task: Data<StatsTask>,
        pg_pool: Data<PgPool>,
    ) {
        eprintln!("Got task: {:?}", task);
        todo!()
    }

    #[sqlx::test]
    pub async fn test(pg_pool: PgPool) {
        let mut mon = Monitor::new();
        let cr = Schedule::from_str("1 * * * * *").unwrap();
        let cron = CronStream::new(cr);
        PostgresStorage::setup(&pg_pool).await.unwrap();
        let store = PostgresStorage::new(pg_pool.clone());
        let backend = cron.pipe_to_storage(store);
        let be = mon
            .register(
                WorkerBuilder::new("stats")
                    .data(StatsTask {
                        warehouse_ident: Uuid::new_v4(),
                    })
                    .data(pg_pool)
                    .backend(backend)
                    .build_fn(collect_stats),
            )
            .run()
            .await
            .unwrap();
    }
}

[twuebi]

With that working, I was curious what apalis is storing in the DB, from what I understand, neither cron schedule nor the task metadata are persisted in the DB and would have to be restored from elsewhere when restarting. I think that also relates to #288, when running in a distributed setup with e.g. 3 replicas, I'd need to somehow ensure that only one replica gets to execute the cron.

[geofmureithi]

In this case, apalis would store:

args: chrono::DateTime<Utc>,

Basically replication should work under the hood, as Postgres guarantees one run.

[twuebi]

That only applies to the next tick, right? Apalis currently doesn't persist the cron schedules somewhere, correct?

[geofmureithi]

Yeah it doesn’t persist the shedules anywhere. You could achieve that using pg_cron.