Scheduled Jobs

The engine runs background work on a schedule: jobs with a standard 5-field cron schedule (minute hour day-of-month month day-of-week, e.g. */5 * * * * for every five minutes, 0 3 * * * for 03:00 every day; ranges 1-5, steps */15, lists 1,15, names jan/mon). Times are wall-clock times in the instance's local time zone.

Who drives the clock

The engine owns no thread: the host calls the scheduler's tick periodically with the current time, and the scheduler runs the jobs that are due.

The CLI (a script or the REPL on a store) and the Python binding do not tick: a store opened there runs no jobs (its workload statistics are still written when it closes). Open the store with the dev server, or tick a Scheduler in your own program, to run them.

Daylight saving time. A schedule is read in the local wall clock. A time that does not exist on the day the clocks go forward (02:30 in most of Europe) runs right after the gap (03:00); a time that happens twice when they go back runs once, at its first occurrence.

A job never overlaps itself. A time missed while nothing ticked (the process was stopped) or while the job ran is skipped: the job runs at its next scheduled time. A failing job (an error, a panic) is recorded and runs again at its next time.

Requests go on while a job runs. The dev server does a job's work on a thread of its own: its session thread, which answers every request, only decides what is due and takes the result back, each a short step. Jobs of one type (query, checkpoint, backup) never run at the same time: a due job waits for the one running and starts at the first tick after it, its time is not missed. Other jobs run beside it. In the browser playground there is no such thread: a job runs in the engine's worker and its requests wait.

While a job of a user runs, the operations of the Store menu that move, remove or write the store's files (checkpoint, backup, rollback, attic restore and remove, compact, closing a database) answer 409 and change nothing: try again when it has finished. What the graph itself cannot offer, a job does not offer either: a query job that writes holds the graph's write lock from its first write until it commits, as any query does, and the system job compact holds it while it runs, so other queries wait for that time. Keep such jobs short with a timeout. A job whose saved query only reads holds no lock between its traversals, so writes go on beside it; each traversal sees the graph as it is at that moment.

Limits per run

A job may limit each of its runs: a timeout and a memory limit (JobLimits, both optional). The engine cannot interrupt arbitrary code, so the limits reach the checks every traversal already makes: while the job runs, its remaining time caps each traversal's evaluationTimeout and its memory limit (bytes) caps each traversal's per-query limit traversal.max_materialized_bytes (a query's own stricter value stays). The memory limit is compared with what the traversal itself holds at once (live bytes, the measure of the per-query limit: memory it frees is given back, what it adds to the graph does not count), never with the memory of the whole process: the process memory ceiling stays the same while a job runs, so a job with 256 MiB still runs in a process that holds 2 GiB. A breach fails the traversal as any timeout or memory limit does, its work is rolled back, and the run is recorded as failed with the reason. An action doing long work of its own checks limits.timeout itself; if it finishes past its timeout anyway, the run is recorded as over time.

System jobs

A Store brings its system jobs (Store::system_jobs()):

System jobs cannot be removed; their schedule and switch can be changed (the playground's and the dev server's job list, POST /api/jobs/schedule and /api/jobs/enable, or WorkloadStats::set_job_override in your own program). The change is a setting of the instance, not of the graph: it is kept in the store's workload sidecar (at the level persisted, the default), so it survives a restart but is not carried by a backup or a fork. Every run (start, duration, error) is recorded in the workload statistics.

Jobs of users

A user's job is a catalog definition (kind job, beside saved queries, compression rules and property indexes): it is versioned with the graph, written to the journal, carried by checkpoints, backups and the catalog file. Its action is one of

g.define_query(#{name: "purge", params: #{label: "string"},
    body: "fn purge(label) { g.v().has_label(label).drop() }"});
g.define_job(#{name: "nightly_purge", schedule: "0 3 * * *", action: "query",
    query: "purge", args: #{label: "software"}, timeout_ms: 30000});
g.jobs().map(|j| [j.name, j.schedule, j.action])

g.define_job(#{name, schedule, action, query?, args?, path?, enabled?, timeout_ms?, memory_limit?, run_on?, description?}) defines or replaces a job, g.jobs() lists them, g.drop_job(name) removes one. The arguments are checked against the query's parameters when the job is defined.

Rights. A job runs with the rights of whoever defined it. When it is defined, the definer's authorizer is asked for everything the job will do: Execute on the saved query and every data write the query may make, Create on the backup directory; the allowed requests are stored with the job as its effects, and every run may do exactly them (and read). If the query is changed later to write more, its runs are refused until the job is defined again. Revoking the definer's rights later does not stop a job until it is redefined (the engine has no identities; an embedding server that has them re-checks on every run). A browser playground cannot define a backup job: its pages have no files.

A job of a newer version (an action this version does not know) is kept as it was written and never run; g.jobs() lists it with served: false and the reason.

Running. The jobs of a store's catalog run on the store's scheduler beside its system jobs: the dev server and the playground pick up a defined, changed or dropped job at their next tick. A query job's run is one transaction whose commit metadata carries job: <name> (the change feed and the history name the job), its queries count under the channel job of the workload statistics, and its limits per run apply. A run that fails (a write outside its effects, a timeout, an error of the query) changes nothing and is recorded with its error; the job runs again at its next time. A job on a graph outside a store is kept but not run (there is no scheduler). In your own program, give the scheduler the store's user jobs and sync them whenever the catalog's jobs changed. Store::user_jobs lists the job definitions it does not schedule beside them, each with the reason (an action or run_on this version does not know, run_on: "replica" on a single instance, a stored schedule that does not parse), so none is dropped silently:

let found = store.user_jobs()?;
for skipped in &found.not_scheduled {
    eprintln!("job {} is not scheduled: {}", skipped.name, skipped.reason);
}
scheduler.sync_user_jobs(found.jobs);

In the playground, Catalog ▸ Scheduled jobs lists every job (the system jobs first) with its schedule, the next and the last run (with its error), the runs of the last two days, a running badge while a run is under way, and a switch; a job of yours has Run now, Edit and Drop, a system job Run now and Schedule (empty: its built-in schedule again). New job opens the form: the name, what it does (a saved query with its arguments as JSON, a checkpoint, a backup directory), the schedule (5-field cron, with presets), on or off, a timeout and a description; a refused job shows the engine's message next to the field.

History. Every run (start, duration, error) is counted in the workload statistics as series job per minute, hour and day: the job list shows the runs of the last two days, Optimize ▸ Workload ▸ Jobs draws them over time, GET /api/jobs/history?name=..&tier=.. answers the buckets.

Front ends. The dev server and the playground share one job API (graphersal-session):

The dev server's user owns the machine's files, so a backup job defined there may name any directory; the browser playground has no files and refuses backup jobs.

In your own program

Your own thread stays free

Scheduler::tick does the work on the calling thread. When that thread must stay free, split it: start_due(now) does the bookkeeping and hands out one JobRun per due job (start_now(name, now) for one on demand), JobRun::run() does the work on any thread, Scheduler::finish takes the result back. Until then the job and its group count as running; Job::with_group puts jobs that must not run together into one group (Store::user_jobs does it by type). A run that cannot be started (no thread) is given up with JobRun::abandon(reason), a failed run like any other.

for run in scheduler.start_due(now) {
    let results = results.clone();
    std::thread::spawn(move || results.send(run.run()));
}
// ... later, on the thread that owns the scheduler:
for finished in finished_runs.try_iter() {
    scheduler.finish(finished);
}

The whole program

use graphersal::jobs::{Job, JobKind, JobLimits, Scheduler};
use graphersal::persist::{MemDir, Store, StoreOptions};

let dir = MemDir::named("memory:jobs");
let store = Store::create(&dir, StoreOptions::new()).unwrap();
let mut scheduler = Scheduler::new()
    .with_time_zone("Europe/Berlin")
    .unwrap()
    .with_stats(store.workload_stats());
for job in store.system_jobs() {
    scheduler.add(job).unwrap();
}
scheduler
    .add(
        Job::new("hello", JobKind::User, "0 * * * *".parse().unwrap(), |run| {
            let _started = run.started_at; // your work; traversals here get the limits below
            Ok(())
        })
        .with_limits(JobLimits::new().with_timeout(std::time::Duration::from_secs(30))),
    )
    .unwrap();
let now = 1_800_000_000_000_000; // microseconds since the Unix epoch
scheduler.tick(now); // plans the first runs; call it again every few seconds
assert_eq!(scheduler.jobs().len(), 3);