Periodic and cron jobs

Periodic jobs make it easy to enqueue a job regularly on a fixed interval or even a complex cron schedule. Periodic job insertion is performed by the cluster’s leader, which holds the schedule in memory. This makes periodic jobs prone to some caveats as the schedule resets across restarts or elections.

See also durable periodic jobs, whose run times are persisted to the database. This is a Pro feature added in River Pro v0.15.

Basic usage

Configure a schedule and a constructor for each periodic job. These examples enqueue a job every fifteen minutes, and once immediately whenever the client becomes leader. Worker clients also need their workers and queues configured.

Periodic jobs are configured as part of the Client. Each job must be initialized with NewPeriodicJob and added to a slice of []*river.PeriodicJob. Each job is defined with a schedule, a constructor, and options.

The following code configures an empty MyPeriodicJobArgs{} job to be inserted every 15 minutes:

periodicJobs := []*river.PeriodicJob{
    river.NewPeriodicJob(
        river.PeriodicInterval(15*time.Minute),
        func() (river.JobArgs, *river.InsertOpts) {
            return MyPeriodicJobArgs{}, nil
        },
        &river.PeriodicJobOpts{RunOnStart: true},
    ),
}

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    PeriodicJobs: periodicJobs,
    // ...
})

See the PeriodicJob example for complete code.

Note the use of the RunOnStart: true option. This option causes the job to be inserted immediately anytime a new leader is elected.

Build a PeriodicJob with an interval and a block that produces job args, then add it to the client configuration:

periodic_job = River::PeriodicJob.new(
  id: :my_periodic_job,
  schedule: River::PeriodicInterval.new(15 * 60),
  run_on_start: true
) { MyPeriodicJobArgs.new }

config = River::Config.new(periodic_jobs: [periodic_job])

Pair an IntervalSchedule with a closure that produces job args, then register the periodic job on the client builder:

use riverqueue::{IntervalSchedule, PeriodicJob, PeriodicJobOpts};
use std::time::Duration;

let periodic_job = PeriodicJob::with_options(
    IntervalSchedule::new(Duration::from_secs(15 * 60))?,
    || MyPeriodicJobArgs {},
    PeriodicJobOpts::new().with_id("my_periodic_job").with_run_on_start(true),
);

let client = riverqueue::Client::builder(pool).periodic_job(periodic_job).build()?;

Add a periodicJob definition to the client’s periodicJobs list. This job runs on a fifteen-minute interval, with an insertion requested on startup too:

import { periodicJob } from "riverqueue";

const client = new Client(driver, {
  periodicJobs: [periodicJob({
    id: "my_periodic_job",
    job: myPeriodicJob,
    args: {},
    every: { minutes: 15 },
    runOnStart: true,
  })],
  workers,
  queues: { default: { maxWorkers: 10 } },
});

Complex cron schedules

Pin the time zone explicitly when clients in different languages or environments share leadership. Ruby defaults to UTC; Go, Rust, and TypeScript default to the process’s local time zone. Five-field cron expressions with CRON_TZ=UTC are portable across all four.

For schedules that are more complex or which require precise control over runtimes, we recommend the robfig/cron package:

go get github.com/robfig/cron/v3

The cron package’s Schedule interface is the same as River’s PeriodicSchedule, meaning that you can use any schedule generated by that package directly in River.

cron.ParseStandard("0 * * * *") // every hour on the hour
cron.ParseStandard("30 * * * *") // every hour on the half hour
cron.ParseStandard("*/15 * * * *") // every 15th minute of every hour, i.e. :00, :15, :30, :45
cron.ParseStandard("CRON_TZ=UTC @midnight") // midnight every night, UTC
cron.ParseStandard("CRON_TZ=America/Chicago @midnight") // midnight every night, in Chicago

Here’s an example that runs MyPeriodicJobArgs{} every 15th minute:

schedule, err := cron.ParseStandard("*/15 * * * *")
if err != nil {
    panic("invalid cron schedule")
}

periodicJobs := []*river.PeriodicJob{
    river.NewPeriodicJob(
        schedule,
        func() (river.JobArgs, *river.InsertOpts) {
            return MyPeriodicJobArgs{}, nil
        },
        nil,
    ),
}

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    PeriodicJobs: periodicJobs,
    // ...
})

See the CronJob example for complete code.

Add gem "fugit" to the Gemfile for cron support:

periodic_job = River::PeriodicJob.new(
  id: :my_periodic_job,
  schedule: River::PeriodicCron.new("*/15 * * * *", timezone: "UTC")
) { MyPeriodicJobArgs.new }

Parse a cron expression with CronSchedule and use it as the periodic job’s schedule. This expression selects every fifteen minutes in UTC:

let periodic_job = riverqueue::PeriodicJob::new(
    riverqueue::CronSchedule::parse("CRON_TZ=UTC */15 * * * *")?,
    || MyPeriodicJobArgs {},
);

Enable River’s chrono-tz Cargo feature for named zones such as America/Chicago.

Use cron to turn a cron expression into a schedule for periodicJob. The explicit UTC zone keeps its timing consistent across deployments:

import { cron, periodicJob } from "riverqueue";

const scheduled = periodicJob({
  id: "my_periodic_job",
  job: myPeriodicJob,
  args: {},
  schedule: cron("CRON_TZ=UTC */15 * * * *"),
});

Adding periodic jobs after client start

River supports adding or removing periodic jobs after client start through the use of Client.PeriodicJobs. For example:

riverClient, err := river.NewClient(...)
if err != nil {
    panic(err)
}

if err := riverClient.Start(ctx); err != nil {
    panic(err)
}

// Add a periodic job after client has already started.
periodicJobHandle := riverClient.PeriodicJobs().Add(
    river.NewPeriodicJob(
        river.PeriodicInterval(15*time.Minute),
        func() (river.JobArgs, *river.InsertOpts) {
            return MyPeriodicJobArgs{}, nil
        },
        nil,
    ),
)

Add a periodic job through the running client’s collection and keep the returned handle if you’ll need to remove it later:

periodic_job_handle = client.periodic_jobs.add(periodic_job)

Call add on the periodic job collection to register a new schedule and obtain its removal handle:

let periodic_job_handle = client.periodic_jobs().add(periodic_job)?;

Add the definition to client.periodicJobs and save the returned handle for later removal:

const periodicJobHandle = client.periodicJobs.add(scheduled);

Adding or removing a job interrupts a running periodic enqueuer service’s wait loop, causing it to immediately insert a new job if running on start was enabled, and scheduling its first run as appropriate.

Adding or removing periodic jobs has no effect unless the client is a cluster’s elected leader, so to guarantee an operation has its desired effect, periodic jobs should be added and removed to or from all River clients across all running processes.

Removing jobs with handle

Adding a periodic job returns a periodic job “handle”, which can later be used to remove the job if necessary:

// Remove a periodic job using handle return by Client.PeriodicJobs.Add.
riverClient.PeriodicJobs().Remove(periodicJobHandle)

Pass the handle returned by add to remove the periodic schedule:

client.periodic_jobs.remove periodic_job_handle

Use the handle returned when the periodic job was added to remove its schedule:

client.periodic_jobs().remove(periodic_job_handle);

Remove the periodic schedule by passing its saved handle to client.periodicJobs.remove:

client.periodicJobs.remove(periodicJobHandle);

Details and caveats

Periodic jobs use the leader election system to ensure that only one worker is managing periodic jobs at any time (in a given database and schema). Periodic jobs are stateless, meaning there is no coordinated or persisted state between workers as far as when the next jobs will run. After a leader election (such as when the previous leader terminates), the new leader will evaluate all the schedules it knows about starting at the current time. It will also insert jobs configured to run on start.

With this architecture, there is a possibility that periodic jobs will sometimes be skipped. River’s leader election is fast, but for a job that only runs at midnight every night, it’s possible that the current leader could be shut down at 11:59:59.99 and the new leader may not take over until 12:00:00.05.

Fortunately, many of these concerns can be addressed by combining periodic jobs with unique jobs and the option to run on start. For example, a job which is configured to be unique at the hourly level will only enqueue once in that hour no matter how many times it’s attempted. By also configuring it to run on start, the above scenario would no longer result in a skipped job.

As with other maintenance processes, insert-only clients do not participate in leader election and are not involved in periodic jobs. To ensure consistent enqueueing of periodic jobs, the same periodic job configuration should be added to all clients eligible to become leader in a given database and schema, including clients written in other languages. Otherwise schedules can disappear when leadership changes. Alternatively, disable leader election on clients that do not have the full schedule.

Alternatively, users can upgrade to River Pro and use durable periodic jobs.

Periodic job metadata

Periodic jobs are inserted with a "periodic": true metadata key, making it possible to distinguish them from other non-periodic jobs in the database.