micropelago/garage
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

lifecycle worker: mitigate potential bugs + refactoring

Browse Source
This commit is contained in:
Alex Auvolat 2023-08-31 00:28:37 +02:00
parent be03a4610f
commit 1cfcc61de8
1 changed files with 31 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
src/model/s3/lifecycle_worker.rs
View File

@ -197,16 +197,21 @@ impl Worker for LifecycleWorker {
async fn wait_for_work(&mut self) -> WorkerState { async fn wait_for_work(&mut self) -> WorkerState {
match &self.state { match &self.state {
State::Completed(d) => { State::Completed(d) => {
let now = now_msec(); let next_day = d.succ_opt().expect("no next day");
let next_start = midnight_ts(d.succ_opt().expect("no next day")); let next_start = midnight_ts(next_day);
if now < next_start { loop {
tokio::time::sleep_until( let now = now_msec();
(Instant::now() + Duration::from_millis(next_start - now)).into(), if now < next_start {
) tokio::time::sleep_until(
.await; (Instant::now() + Duration::from_millis(next_start - now)).into(),
)
.await;
} else {
break;
}
} }
self.state = State::Running { self.state = State::Running {
date: today(), date: std::cmp::max(next_day, today()),
pos: vec![], pos: vec![],
counter: 0, counter: 0,
objects_expired: 0, objects_expired: 0,
@ -228,6 +233,14 @@ async fn process_object(
mpu_aborted: &mut usize, mpu_aborted: &mut usize,
last_bucket: &mut Option<Bucket>, last_bucket: &mut Option<Bucket>,
) -> Result<Skip, Error> { ) -> Result<Skip, Error> {
if !object
.versions()
.iter()
.any(|x| x.is_data() || x.is_uploading(None))
{
return Ok(Skip::NextObject);
}
let bucket = match last_bucket.take() { let bucket = match last_bucket.take() {
Some(b) if b.id == object.bucket_id => b, Some(b) if b.id == object.bucket_id => b,
_ => garage _ => garage
@ -276,7 +289,7 @@ async fn process_object(
if let Ok(exp_date) = parse_lifecycle_date(&exp_date) { if let Ok(exp_date) = parse_lifecycle_date(&exp_date) {
now_date >= exp_date now_date >= exp_date
} else { } else {
warn!("Invalid expiraiton date stored in bucket {:?} lifecycle config: {}", bucket.id, exp_date); warn!("Invalid expiration date stored in bucket {:?} lifecycle config: {}", bucket.id, exp_date);
false false
} }
} }
@ -309,17 +322,15 @@ async fn process_object(
.iter() .iter()
.filter_map(|v| { .filter_map(|v| {
let version_date = next_date(v.timestamp); let version_date = next_date(v.timestamp);
match &v.state { if (now_date - version_date) >= chrono::Duration::days(*abort_mpu_days as i64)
ObjectVersionState::Uploading { .. } && matches!(&v.state, ObjectVersionState::Uploading { .. })
if (now_date - version_date) {
>= chrono::Duration::days(*abort_mpu_days as i64) => Some(ObjectVersion {
{ state: ObjectVersionState::Aborted,
Some(ObjectVersion { ..*v
state: ObjectVersionState::Aborted, })
..*v } else {
}) None
}
_ => None,
} }
}) })
.collect::<Vec<_>>(); .collect::<Vec<_>>();