use std::collections::HashMap; use std::sync::{Arc, Mutex as StdMutex}; use std::time::Duration; use tokio::sync::Mutex; use crate::storage_client::StorageClient; use super::pull::sync_pull; use super::push::sync_push; use super::store::{unix_now, Direction, Entry, Store as MirrorStore}; /// How often the scheduler checks for due mirrors. const TICK_INTERVAL: Duration = Duration::from_secs(30); /// Periodically syncs configured mirrors whose interval has elapsed. pub struct Scheduler { storage: Arc, mirrors: Arc, locks: StdMutex>>>, } impl Scheduler { pub fn new(storage: Arc, mirrors: Arc) -> Arc { Arc::new(Scheduler { storage, mirrors, locks: StdMutex::new(HashMap::new()) }) } /// Ticks forever. pub async fn run(self: Arc) { let mut ticker = tokio::time::interval(TICK_INTERVAL); loop { self.tick().await; ticker.tick().await; } } async fn tick(self: &Arc) { let entries = match self.mirrors.all().await { Ok(e) => e, Err(e) => { eprintln!("mirror scheduler: load mirrors: {e}"); return; } }; let now = unix_now(); for entry in entries { let due = entry.last_sync_unix + entry.interval_seconds; if now < due { continue; } let this = self.clone(); tokio::spawn(async move { this.sync(entry).await }); } } async fn sync(self: Arc, entry: Entry) { let lock = self.lock_for(&entry); let Ok(_guard) = lock.try_lock() else { return; // a sync for this mirror is already running }; let result = match entry.direction { Direction::Pull => sync_pull(&self.storage, &entry).await, Direction::Push => sync_push(&self.storage, &entry).await, }; if let Err(e) = result { eprintln!("mirror sync {:?} {}: {e}", entry.direction, entry.repo); } if let Err(e) = self.mirrors.touch(&entry, unix_now()).await { eprintln!("mirror sync {:?} {}: record last sync: {e}", entry.direction, entry.repo); } } fn lock_for(&self, entry: &Entry) -> Arc> { let mut locks = self.locks.lock().unwrap(); locks.entry(format!("{:?}:{}", entry.direction, entry.repo)).or_insert_with(|| Arc::new(Mutex::new(()))).clone() } }