yolo/arr
3
0
Fork 0
Code Issues 12 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(daemon): stagger reconcile lanes

Browse Source
This commit is contained in:
Miguel Palhas
2026-08-22 21:32:05 +01:00
parent 79516eedec
commit d541ea0d98
1 changed files with 32 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/arr-daemon/src/reconcile.rs
+32 -15
View File
@@ -16,6 +16,11 @@ pub const RSS_INTERVAL: Duration = Duration::from_mins(10);
pub const METADATA_INTERVAL: Duration = Duration::from_hours(24);
pub const REAPER_INTERVAL: Duration = Duration::from_mins(5);
const RECONCILE_START_DELAY: Duration = Duration::ZERO;
const METADATA_START_DELAY: Duration = Duration::from_secs(5);
const REAPER_START_DELAY: Duration = Duration::from_secs(10);
const RSS_START_DELAY: Duration = Duration::from_secs(20);
pub type ActionError = Box<dyn Error + Send + Sync>;
pub type ActionFuture<'a> =
Pin<Box<dyn Future<Output = Result<Vec<Outcome>, ActionError>> + Send + 'a>>;
@@ -102,6 +107,15 @@ impl Schedule {
Tick::Reaper => self.reaper,
}
}
const fn start_delay(tick: Tick) -> Duration {
match tick {
Tick::Reconcile => RECONCILE_START_DELAY,
Tick::Rss => RSS_START_DELAY,
Tick::Metadata => METADATA_START_DELAY,
Tick::Reaper => REAPER_START_DELAY,
}
}
}
/// The daemon's four-lane scheduler. Actions register into a lane; no queued
@@ -140,41 +154,36 @@ impl ReconcileLoop {
/// actions can reconstruct transient state before serving later ticks.
pub async fn run(self, shutdown: watch::Receiver<bool>) {
let this = Arc::new(self);
let reconcile = Arc::clone(&this).run_lane(Tick::Reconcile, true, shutdown.clone());
let rss = Arc::clone(&this).run_lane(Tick::Rss, false, shutdown.clone());
let metadata = Arc::clone(&this).run_lane(Tick::Metadata, false, shutdown.clone());
let reaper = Arc::clone(&this).run_lane(Tick::Reaper, false, shutdown);
let reconcile = Arc::clone(&this).run_lane(Tick::Reconcile, shutdown.clone());
let rss = Arc::clone(&this).run_lane(Tick::Rss, shutdown.clone());
let metadata = Arc::clone(&this).run_lane(Tick::Metadata, shutdown.clone());
let reaper = Arc::clone(&this).run_lane(Tick::Reaper, shutdown);
tokio::join!(reconcile, rss, metadata, reaper);
}
async fn run_lane(
self: Arc<Self>,
tick: Tick,
run_immediately: bool,
mut shutdown: watch::Receiver<bool>,
) {
async fn run_lane(self: Arc<Self>, tick: Tick, mut shutdown: watch::Receiver<bool>) {
if *shutdown.borrow() {
return;
}
let interval = self.schedule.interval(tick);
let first_tick = if run_immediately {
Instant::now()
} else {
Instant::now() + interval
};
let first_tick = Instant::now() + Schedule::start_delay(tick);
let mut timer = tokio::time::interval_at(first_tick, interval);
timer.set_missed_tick_behavior(MissedTickBehavior::Skip);
loop {
tokio::select! {
biased;
changed = shutdown.changed() => {
if changed.is_err() || *shutdown.borrow() {
return;
}
}
_ = timer.tick() => {
if *shutdown.borrow() {
return;
}
self.run_tick(tick).await;
}
}
@@ -356,6 +365,14 @@ mod tests {
assert_eq!(schedule.interval(Tick::Rss), Duration::from_mins(10));
assert_eq!(schedule.interval(Tick::Metadata), Duration::from_hours(24));
assert_eq!(schedule.interval(Tick::Reaper), Duration::from_mins(5));
assert_eq!(Schedule::start_delay(Tick::Reconcile), Duration::ZERO);
assert_eq!(
Schedule::start_delay(Tick::Metadata),
Duration::from_secs(5)
);
assert_eq!(Schedule::start_delay(Tick::Reaper), Duration::from_secs(10));
assert_eq!(Schedule::start_delay(Tick::Rss), Duration::from_secs(20));
}
#[tokio::test]