Menu
Public repository / 8d41f1e10aceed387d425b43038f2070e399a349
AkurAI-Build / src/schedule.rs
Latest change 8d41f1e10aceed387d425b43038f2070e399a349 - OLI-277: schedules, reusable workflows (include + inputs), per-repository secrets (value or PassVault), native caches by Ólafur Búi Ólafsson
//! Scheduled runs (OLI-277): every minute the forge reads each hosted
//! repository's pipeline on its default branch, and queues a run for every
//! `schedules` entry whose cron matches that minute, once per minute per
//! schedule (the `schedule_fires` table makes a restart idempotent).
use std::{path::Path, process::Command};
use anyhow::{Context, Result};
use rusqlite::params;
use serde::Serialize;
use crate::{
config::{self, Loader, Schedule},
cron::Cron,
db::Database,
hosted_git,
runner::Runner,
};
/// Reads pipelines straight from hosted bare repositories.
pub struct HostedLoader<'a> {
pub root: &'a Path,
pub database: &'a Database,
/// The repository whose pipeline is being read; `include` without `repo` means this one.
pub repository: &'a str,
pub reference: &'a str,
}
impl Loader for HostedLoader<'_> {
fn load(&self, repo: Option<&str>, reference: Option<&str>, path: &str) -> Result<String> {
let (repo, reference) = match repo {
None => (self.repository.to_owned(), self.reference.to_owned()),
Some(repo) => {
let repository = self.database.repository(repo)?;
(
repo.to_owned(),
reference.unwrap_or(&repository.default_branch).to_owned(),
)
}
};
read_file(self.root, &repo, &reference, path)
}
}
/// `git show <reference>:<path>` in the hosted mirror; `Ok(None)` when absent.
pub fn read_file_optional(
root: &Path,
repository: &str,
reference: &str,
path: &str,
) -> Result<Option<String>> {
let bare = hosted_git::repository_path(root, repository)?;
if !bare.is_dir() {
return Ok(None);
}
let output = Command::new("git")
.arg("-C")
.arg(&bare)
.args(["show", &format!("{reference}:{path}")])
.env("GIT_TERMINAL_PROMPT", "0")
.output()
.context("read a file from the hosted repository")?;
if !output.status.success() {
return Ok(None);
}
Ok(Some(
String::from_utf8(output.stdout).context("file is not UTF-8")?,
))
}
fn read_file(root: &Path, repository: &str, reference: &str, path: &str) -> Result<String> {
read_file_optional(root, repository, reference, path)?
.with_context(|| format!("{repository}@{reference}:{path} is not readable"))
}
/// The schedules a hosted repository declares on its default branch.
pub fn schedules_of(root: &Path, database: &Database, repository: &str) -> Result<Vec<Schedule>> {
let repository = database.repository(repository)?;
let source = match read_file_optional(
root,
&repository.name,
&repository.default_branch,
".akurai.yml",
)? {
Some(source) => source,
None => return Ok(Vec::new()),
};
let loader = HostedLoader {
root,
database,
repository: &repository.name,
reference: &repository.default_branch,
};
Ok(config::parse_pipeline(&source, &loader)?.schedules)
}
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct Fired {
pub repository: String,
pub branch: String,
pub cron: String,
pub run_id: i64,
}
/// Queue every schedule due in the minute containing `now`. Repositories whose
/// pipeline does not parse are skipped (their next run will report the error).
pub fn tick(database: &Database, runner: &Runner, root: &Path, now: i64) -> Result<Vec<Fired>> {
let minute = now.div_euclid(60);
let mut fired = Vec::new();
for repository in database.repositories()? {
let schedules = match schedules_of(root, database, &repository.name) {
Ok(schedules) => schedules,
Err(error) => {
tracing::warn!(repository = %repository.name, %error, "pipeline schedules unreadable");
continue;
}
};
for schedule in schedules {
let cron = match Cron::parse(&schedule.cron) {
Ok(cron) => cron,
Err(_) => continue,
};
if !cron.matches(minute * 60) {
continue;
}
let branch = schedule
.branch
.clone()
.unwrap_or_else(|| repository.default_branch.clone());
let first = database.connection()?.execute(
"INSERT OR IGNORE INTO schedule_fires(repository_id, cron, branch, minute) VALUES (?1, ?2, ?3, ?4)",
params![repository.id, schedule.cron, branch, minute],
)?;
if first == 0 {
continue;
}
let run_id = runner.queue(&repository.name, Some(&branch), None, "manual")?;
database.connection()?.execute(
"UPDATE runs SET schedule = ?2 WHERE id = ?1",
params![run_id, schedule.cron],
)?;
database.connection()?.execute(
"UPDATE schedule_fires SET run_id = ?5 WHERE repository_id = ?1 AND cron = ?2 AND branch = ?3 AND minute = ?4",
params![repository.id, schedule.cron, branch, minute, run_id],
)?;
fired.push(Fired {
repository: repository.name.clone(),
branch,
cron: schedule.cron.clone(),
run_id,
});
}
}
// Keep two days of fire history.
database.connection()?.execute(
"DELETE FROM schedule_fires WHERE minute < ?1",
[minute - 2 * 24 * 60],
)?;
Ok(fired)
}
#[cfg(test)]
mod tests {
use super::*;
use std::fs;
const KEY: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
fn git(dir: &Path, args: &[&str]) {
let status = Command::new("git")
.arg("-C")
.arg(dir)
.args(args)
.status()
.expect("git");
assert!(status.success(), "git {args:?}");
}
#[test]
fn due_schedules_queue_one_run_per_minute_and_survive_a_rerun() -> Result<()> {
let temp = tempfile::tempdir()?;
let root = temp.path();
let database = Database::memory(KEY)?;
let work = root.join("work");
fs::create_dir_all(&work)?;
git(&work, &["init", "-q", "-b", "main"]);
git(&work, &["config", "user.name", "T"]);
git(&work, &["config", "user.email", "t@example"]);
fs::create_dir_all(work.join(".akurai"))?;
fs::write(
work.join(".akurai/shared.yml"),
"version: 1\njobs:\n - name: shared\n executor: native\n run: echo ${{ inputs.what }}\n",
)?;
fs::write(
work.join(".akurai.yml"),
"version: 1\nschedules:\n - cron: '30 18 * * *'\n - cron: '*/5 * * * *'\n branch: nightly\ninclude:\n - path: .akurai/shared.yml\n with: {what: hi}\njobs:\n - name: build\n executor: native\n run: echo ok\n",
)?;
git(&work, &["add", "."]);
git(&work, &["commit", "-q", "-m", "pipeline"]);
let hosted = hosted_git::host(root, "demo", &work)?;
database.add_repository("demo", hosted.to_str().expect("utf8"), "main")?;
database.add_repository("elsewhere", "https://example.com/x.git", "main")?;
let runner = Runner::new(database.clone(), root.to_owned(), true)?;
let found = schedules_of(root, &database, "demo")?;
assert_eq!(found.len(), 2);
assert!(
schedules_of(root, &database, "elsewhere")?.is_empty(),
"not hosted: nothing"
);
let at = 1_791_052_200; // 2026-10-03 18:30 UTC: both schedules due
let fired = tick(&database, &runner, root, at + 17)?;
assert_eq!(fired.len(), 2, "{fired:?}");
assert_eq!(fired[0].branch, "main");
assert_eq!(fired[1].branch, "nightly");
let run = database.run(fired[0].run_id)?;
assert_eq!(run.schedule.as_deref(), Some("30 18 * * *"));
assert_eq!(run.trigger, "manual");
assert!(
tick(&database, &runner, root, at + 40)?.is_empty(),
"same minute: nothing new"
);
let later = tick(&database, &runner, root, at + 60)?;
assert!(later.is_empty(), "18:31 matches neither");
let five = tick(&database, &runner, root, at + 300)?;
assert_eq!(five.len(), 1);
assert_eq!(five[0].cron, "*/5 * * * *");
Ok(())
}
}