use super::lease;
use super::{LeanError, Syncer};
pub fn env_list(name: &str) -> Option<Vec<String>> {
let raw = std::env::var(name).ok()?;
let v: Vec<String> =
raw.split(',').map(|e| e.trim().to_string()).filter(|e| !e.is_empty()).collect();
if v.is_empty() {
return None;
}
Some(v)
}
pub enum Step {
Checkout,
Barrier,
Sync,
Rescope(Option<Vec<String>>),
}
impl Step {
pub fn holds_the_fence_throughout(&self) -> bool {
match self {
Step::Rescope(_) => true,
Step::Checkout | Step::Barrier | Step::Sync => false,
}
}
}
pub async fn run_verb(sc: &mut Syncer, step: Step) -> Result<(), LeanError> {
match &step {
Step::Barrier => sc.refuse_if_read("barrier")?,
Step::Rescope(_) => sc.refuse_if_read("rescope")?,
Step::Checkout | Step::Sync => {}
}
if step.holds_the_fence_throughout() {
claim_then(sc, step).await
} else {
lease_free_then(sc, step).await
}
}
pub async fn claim_then(sc: &mut Syncer, step: Step) -> Result<(), LeanError> {
lease::verify_claim(sc).await?;
lease::warn_if_prefix_is_shared(sc).await;
let held = lease::claim(sc).await?;
eprintln!("flint-sync: holding epoch {}", held.epoch);
let out = run_step(sc, step).await;
if let Err(e) = lease::release(sc).await {
eprintln!("flint-sync: the publish fence could not be released ({e}); a waiter deposes it");
}
out
}
pub async fn lease_free_then(sc: &mut Syncer, step: Step) -> Result<(), LeanError> {
lease::verify_claim(sc).await?;
lease::warn_if_prefix_is_shared(sc).await;
run_step(sc, step).await
}
pub async fn run_step(sc: &mut Syncer, step: Step) -> Result<(), LeanError> {
{
match step {
Step::Checkout => {
let r = sc.checkout_scoped(env_list("FLINT_SYNC_CHECKOUT_SCOPE")).await?;
eprintln!(
"flint-sync: checkout — {} materialized, {} present, live-tree={}",
r.materialized, r.skipped_present, r.resumed_live_tree
);
if let Some(sc) = &r.scope {
eprintln!(
"flint-sync: checkout SCOPED to {:?} — {} citations declined",
sc, r.out_of_scope
);
}
eprintln!(
"flint-sync: phase manifest={:.3}s fetch={:.3}s commit={:.3}s bytes={} ranged={}",
r.manifest_secs, r.fetch_secs, r.commit_secs, r.bytes, r.ranged
);
}
Step::Rescope(target) => {
let r = sc.rescope(target).await?;
eprintln!(
"flint-sync: rescope to {:?} — uncited {}, unlinked {}, materialised {} \
({} bytes), already held {}",
r.target, r.uncited, r.unlinked, r.materialized, r.bytes, r.already_held
);
if !r.kept_dirty.is_empty() {
eprintln!(
"flint-sync: rescope KEPT {} path(s) with unpublished changes: {:?} — \
publish them, then rescope again to drop them",
r.kept_dirty.len(),
r.kept_dirty
);
}
}
Step::Barrier => {
let r = sc.run_barrier().await?;
eprintln!(
"flint-sync: barrier seq={:?} up={} del={} parked={} consumed={}{}",
r.seq,
r.uploaded.len(),
r.deleted.len(),
r.parked.len(),
r.consumed,
if r.leaked.is_empty() {
String::new()
} else {
format!(" leaked={}", r.leaked.len())
}
);
}
Step::Sync => {
let r = sc.sync().await?;
println!("{}", serde_json::to_string_pretty(&r).unwrap());
}
}
Ok(())
}
}