Cron jobs

Most "cron" work in a microservice is a scan: expire stale sessions, retry stuck work, reconcile a search index. A CronJob is exactly that: a Processor of CronTicks with a constant INTERVAL, a CronJobInterval that says how its runs are spaced. The interval is a constant of the job type, not a calendar expression: there is no "every day at 03:00" syntax.

If you come from Spring, the closest thing is @Scheduled on a bean method, and the three kinds of interval map onto its options: AfterFinish is fixedDelay, AtLeast is fixedRate without overlapping runs, and AfterStart is fixedRate with @Async. One difference is covered below: every replica of the service runs every job.

Defining a jobCopy Anchor Link

A job is a struct holding its dependencies. It implements three traits:

• ServiceCreation, to be built from a value registered in the service builder;
• Processor<CronTick> with Output = () and an Error that implements Display, which is one run;
• CronJob, which adds the INTERVAL.

The job must also be Send + Sync + 'static, because it runs on its own Tokio task.

use std::time::Duration;

use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::{CronJob, CronJobInterval, CronTick, ServiceCreation};

pub struct ExpireSessions {
    store: SessionStore,
}

impl ServiceCreation for ExpireSessions {
    type Dep = SessionStore;

    fn create_service(store: SessionStore) -> Self {
        Self { store }
    }
}

impl Processor<CronTick> for ExpireSessions {
    type Output = ();
    type Error = StoreError;

    async fn process(&self, _tick: CronTick) -> Result<(), StoreError> {
        let deleted = self.store.delete_expired(1_000).await?;
        tracing::info!(deleted, "expired sessions removed");
        Ok(())
    }
}

impl CronJob for ExpireSessions {
    const INTERVAL: CronJobInterval = CronJobInterval::AtLeast(Duration::from_secs(60));
}

SessionStore and StoreError stand in for your own data access. The store is cheap to clone (it is cloned out of the service builder when the job is built), and the error only needs Display, because that is how a failed run is logged:

/// Stand-in for a database pool: cheap to clone, every clone shares the data.
#[derive(Clone, Default)]
pub struct SessionStore {
    expired: Arc<AtomicU64>,
}

impl SessionStore {
    pub async fn delete_expired(&self, limit: u64) -> Result<u64, StoreError> {
        self.expired.fetch_add(1, Ordering::Relaxed);
        Ok(limit.min(1))
    }
}

#[derive(Debug)]
pub struct StoreError(String);

impl fmt::Display for StoreError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "session store: {}", self.0)
    }
}

If you use the data layer, a Db handle is a natural Dep, and its query errors already implement Display; see Data access.

The tickCopy Anchor Link

Each run receives a CronTick:

run counts runs that actually happened. How far apart they are depends on the interval kind and on how long runs take, and AfterStart does not count grid points it had to skip, so run is not "time since start divided by the interval". It is good for things like "do a full pass every Nth run":

impl Processor<CronTick> for ReindexSearch {
    type Output = ();
    type Error = StoreError;

    async fn process(&self, tick: CronTick) -> Result<(), StoreError> {
        // Every 24th run (and the first one) rebuilds everything;
        // the others only pick up changes.
        let reindexed = if tick.run % 24 == 0 {
            self.index.reindex_all().await?
        } else {
            self.index.reindex_changed().await?
        };
        tracing::info!(run = tick.run, fired_at = ?tick.fired_at, reindexed, "search index refreshed");
        Ok(())
    }
}

impl CronJob for ReindexSearch {
    const INTERVAL: CronJobInterval = CronJobInterval::AfterFinish(Duration::from_secs(60 * 60));
}

Registering and startingCopy Anchor Link

ServiceBuilder::cron_job::<Job, Position>() looks up the job's Dep, clones it, builds the job with create_service and starts a job list. Chain one call per job, then start:

let (services, cron) = ServiceBuilder::new(SessionStore::default())
    .push(SearchIndex::default())
    .cron_job::<ExpireSessions, _>()
    .cron_job::<ReindexSearch, _>()
    .start();

// `services` is handed back: register gRPC servers, consumers, ... from it.
let _store: &SessionStore = services.provide::<SessionStore, _>();

let _ = tokio::signal::ctrl_c().await;
// Lets the runs in progress finish, then stops every job.
cron.shutdown().await;

• Position is _ unless the Dep type is registered more than once (then use Here / There, see The service builder). A Dep that was never registered is a compile error.
• A job that needs several registered values is registered with compose_cron_job instead; see Jobs built from several services.
• start spawns one Tokio task per job and returns the service builder together with a CronRuntime. It must be called inside a Tokio runtime; outside one it panics.
• Jobs are stored in a type-level list, like gRPC servers and the service builder itself: no boxing, no dynamic dispatch.

The scheduleCopy Anchor Link

Every job's first run is at start, the moment start was called, not one interval later. After that, the variant of INTERVAL decides when the next run starts, for its duration d:

The three only differ once a run takes a while. In the diagram, the first run takes 2.5·d and every later run 0.25·d:

Different jobs are independent: a slow ReindexSearch never delays ExpireSessions.

Choosing an intervalCopy Anchor Link

AtLeast(d): at most one run per d, never two at once. This is the usual choice for a scan that should run "about every d". A run that overruns delays the next one instead of piling up a backlog: the next run starts as soon as it finished, then the job is back on its pace. Consecutive starts are at least d apart on the monotonic clock, so their fired_at are too, unless the wall clock is adjusted in between.

AfterFinish(d): a guaranteed pause between runs. The time between two starts is d plus the run time, so the schedule drifts by every run's duration. What you get in exchange is rest for whatever the job puts load on: a database or a rate-limited API always gets d of quiet after a run, however long that run was. Choose it for work that drains a backlog in batches, or whose run time grows with load.

AfterStart(d): a fixed grid, runs may overlap. Every grid point starts a run in its own task, whether the previous run has finished or not. Choose it when timing matters more than exclusivity, such as sampling or a heartbeat that must not slip because one run was slow. Keep in mind:

• The job runs concurrently with itself. process takes &self, so any state it shares between runs needs synchronization, and the work itself must tolerate two runs at once.
• If runs keep taking longer than d, they keep accumulating: nothing bounds how many run at once.
• Grid points missed because the runtime was blocked (for example by blocking code on a worker thread) are skipped, not run in a burst, and run does not count them.
• fired_at includes the wake-up delay, so two consecutive fired_at can be slightly less than d apart. That matters for fences across replicas.

Errors and panicsCopy Anchor Link

A run that returns Err is logged with tracing at error level (cron job failed, with the error's Display output). A run that panics is caught and logged as well (cron job panicked, with the panic message). In both cases the job keeps its schedule: the next run starts when its interval says, as if the run had succeeded.

Every run executes inside a cron_job tracing span with the fields job (the job's type name) and run, so everything your job logs can be traced back to the run. Install a tracing subscriber in main to see these logs.

A zero interval does not buildCopy Anchor Link

A duration of zero, in any of the three variants, has no schedule, so registering such a job is rejected at compile time. Here Busy implements ServiceCreation and Processor<CronTick> as usual:

impl CronJob for Busy {
    const INTERVAL: CronJobInterval = CronJobInterval::AtLeast(Duration::ZERO); // no schedule
}

#[tokio::main]
async fn main() {
    let (_services, cron) = ServiceBuilder::new(Config)
        .cron_job::<Busy, _>() // `cargo build` fails: INTERVAL is zero
        .start();
    cron.shutdown().await;
}

cargo build stops with (abridged):

error[E0080]: evaluation panicked: CronJob::INTERVAL must not be zero
   --> wakuwaku/src/services/cron_job.rs:300:17
    |
300 |         const { assert_interval::<Job>() };
    |                 ^^^^^^^^^^^^^^^^^^^^^^^^ evaluation of `wakuwaku::services::cron_job::CronRegisterCenter::<Busy>::new::{constant#0}` failed inside this call
...
note: the above error was encountered while instantiating `fn CronRegisterCenter::<Busy>::new`

The error points into wakuwaku's source rather than at your cron_job call; the job type in CronRegisterCenter::<Busy> tells you which job to fix.

Running on several replicasCopy Anchor Link

Jobs run locally: every process that starts a job runs it. With three replicas of a service, each run happens three times, and because each replica's schedule starts when that replica started, the three runs are not aligned in time.

Design each job for that. In order of preference:

1. Make it idempotent. "Delete sessions that expired" is safe to run three times; the second and third runs find nothing to do.
2. Partition the work. Have each replica scan only its share of the keys (for example by a hash of the id modulo the replica count).
3. Take a lock inside the run, such as the Redis Lock from the redis feature (see Redis and Locks).

With a lock, every replica tries to take it and at most one succeeds; the others skip that run:

use std::time::Duration;

use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::deps::redis::RedisError;
use wakuwaku::entity::redis::RedisSource;
use wakuwaku::entity::redis::collection::lock::Lock;
use wakuwaku::entity::{Db, ReadWrite};
use wakuwaku::{CronJob, CronJobInterval, CronTick, ServiceCreation};

/// One lock per job name. The lease must outlast a run.
const CRON_LOCKS: Lock<str> = Lock::new("cron", Duration::from_secs(5 * 60));

pub struct SendDailyDigest {
    db: Db<RedisSource, ReadWrite>,
}

impl ServiceCreation for SendDailyDigest {
    type Dep = Db<RedisSource, ReadWrite>;

    fn create_service(db: Db<RedisSource, ReadWrite>) -> Self {
        Self { db }
    }
}

impl Processor<CronTick> for SendDailyDigest {
    type Output = ();
    type Error = RedisError;

    async fn process(&self, _tick: CronTick) -> Result<(), RedisError> {
        // Every replica tries; at most one gets the token.
        let Some(token) = self.db.process(CRON_LOCKS.try_lock("send-daily-digest")).await? else {
            return Ok(()); // Another replica is running it right now.
        };

        // ... build and send the digest; call `token.extend()` for long work ...

        self.db.process(token.release()).await?;
        Ok(())
    }
}

impl CronJob for SendDailyDigest {
    const INTERVAL: CronJobInterval = CronJobInterval::AtLeast(Duration::from_secs(60 * 60));
}

Acquiring never waits: try_lock outputs Some(token) if the lock was free and None otherwise. The lease (the lock's TTL) is what frees the lock if the holder crashes, so work that may outlive it must call token.extend().

Once per interval across replicasCopy Anchor Link

To run each pass about once per d across all replicas, let a run claim the pass in shared storage with a fence on fired_at: the run goes ahead only if it can atomically record its own fired_at and the last recorded one is at least d earlier. In SQL, with one row per job:

-- $1 = job name, $2 = the tick's fired_at.
-- The run goes ahead only if this changes a row
-- (seed each job's row with an old last_fired_at).
UPDATE cron_claims
SET    last_fired_at = $2
WHERE  job = $1
  AND  last_fired_at <= $2 - INTERVAL '1 hour';

The Redis lock can play the same role: do not release it, and set its lease a little shorter than d, so the lock itself records "someone ran this recently". The work must then finish well within the lease.

Use AfterFinish or AtLeast for a fenced job. Their consecutive fired_at on one replica are at least d apart, so a replica never fences out its own next run. AfterStart runs can fire slightly less than d apart, and the fence would then reject a run that should have gone ahead, skipping a whole pass.

Jobs built from several servicesCopy Anchor Link

cron_job builds a job from one registered value, its Dep. A job that needs several registered values, or that you build by hand, is registered with compose_cron_job: the closure borrows the service builder, provides whatever it needs, and returns the job (see Composing components with a closure). The job needs no ServiceCreation impl and is not registered as a service. cron_job and compose_cron_job chain freely:

/// Needs two registered values, so it is composed.
pub struct ReconcileIndex {
    store: SessionStore,
    index: SearchIndex,
}

let (services, cron) = ServiceBuilder::new(SessionStore::default())
    .push(SearchIndex::default())
    .cron_job::<ExpireSessions, _>()
    .compose_cron_job(|s| ReconcileIndex {
        store: s.provide::<SessionStore, _>().clone(),
        index: s.provide::<SearchIndex, _>().clone(),
    })
    .start();

A composed job still implements Processor<CronTick> and CronJob; it only skips ServiceCreation. To start the jobs later, end the chain with into_parts instead of start: it returns the service builder and the unstarted CronRegisterCenter. CronRegisterCenter also works without a service builder at all:

use wakuwaku::services::cron_job::CronRegisterCenter;

let store = SessionStore::default();
let cron = CronRegisterCenter::new(ExpireSessions::create_service(store.clone()))
    .push(ReconcileIndex { store, index: SearchIndex::default() })
    .start();
cron.shutdown().await;

new and push both apply the zero-interval check.

Shutting downCopy Anchor Link

CronRuntime gives you three ways to stop the jobs:

Prefer shutdown, so a job is never cut off in the middle of a batch. shutdown waits as long as the runs take, so a run that hangs would block it forever; bound it with a timeout if that matters:

if tokio::time::timeout(Duration::from_secs(30), cron.shutdown()).await.is_err() {
    // Timing out drops the `CronRuntime`, which aborts the stuck runs.
    tracing::warn!("cron jobs did not stop within 30s; aborted");
}

In a service that also serves gRPC, start the jobs, build the servers from the service builder that start hands back, and stop the jobs after the server has drained:

let (services, cron) = ServiceBuilder::new(AccountStore::default())
    .push(SessionStore::default())
    .cron_job::<ExpireSessions, _>()
    .start();

let (_services, register) = services
    .grpc_service::<AccountsServer<_>, AccountsService, _>(AccountsServer::new)
    .into_router_fn();

let router = Server::builder().add_routes(Routes::default());
register(router)
    .serve_with_shutdown("[::1]:50051".parse()?, async {
        let _ = tokio::signal::ctrl_c().await;
    })
    .await?;

// The server has drained; now let the jobs finish their current run.
cron.shutdown().await;

See gRPC services for the server side and Lifecycle for the order across a whole service.

Testing a jobCopy Anchor Link

A job is a plain value, so a unit test builds it and calls process with a CronTick made by hand. No runtime, no waiting for the interval:

#[cfg(test)]
mod tests {
    use std::time::SystemTime;

    use super::*;

    #[tokio::test]
    async fn expire_sessions_deletes_from_the_store() {
        let store = SessionStore::default();
        let job = ExpireSessions::create_service(store.clone());

        let tick = CronTick { run: 0, fired_at: SystemTime::now() };
        job.process(tick).await.unwrap();

        assert_eq!(store.expired.load(Ordering::Relaxed), 1);
    }
}

To test the scheduling itself, start the real runtime under Tokio's paused clock, where sleep advances virtual time instantly. This needs Tokio's test-util feature in [dev-dependencies] (tokio = { version = "1", features = ["test-util"] }):

#[tokio::test(start_paused = true)]
async fn the_first_run_happens_at_start() {
    let store = SessionStore::default();
    let (_services, cron) = ServiceBuilder::new(store.clone())
        .cron_job::<ExpireSessions, _>()
        .start();

    // Half of the 60-second interval: only the first run has happened.
    tokio::time::sleep(Duration::from_secs(30)).await;
    cron.shutdown().await;

    assert_eq!(store.expired.load(Ordering::Relaxed), 1);
}

Most jobs only need the first kind of test: the schedule is wakuwaku's job and is tested there. See Testing for the general approach.