use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
use super::{LeanError, LeanResult};
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct BaselineEntry {
pub etag: String,
pub generation: u64,
pub size: u64,
pub mtime_unix: i64,
#[serde(default)]
pub mtime_nanos: Option<u32>,
pub crc64_b64: Option<String>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct Baseline {
pub seq: u64,
pub manifest_etag: Option<String>,
pub entries: BTreeMap<String, BaselineEntry>,
pub inst_base: BTreeMap<String, String>,
pub prev_scan: BTreeSet<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ScopeIntent {
pub target: Option<Vec<String>>,
#[serde(default)]
pub drop: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Incarnation {
pub holder_id: String,
pub epoch: u64,
pub last_token: Option<String>,
pub quiet_polls: u32,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct ForeignChange {
pub path: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub etag: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub crc64_b64: Option<String>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct IntentJournal {
pub flush_uuid: String,
pub keys: Vec<String>,
pub recent_uuids: Vec<String>,
#[serde(default)]
pub installed_etag: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub installed_foreign: Vec<ForeignChange>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub declared_deletes: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConflictRecord {
pub path: String,
pub foreign_etag: String,
pub preserved_key: Option<String>,
pub kind: String, pub at_unix: u64,
}
pub struct SyncerState {
dir: PathBuf,
_lock: std::fs::File,
}
const MARKER: &str = "checkout-complete";
pub const DRAINED: &str = "drained.json";
const LOCK: &str = "lock";
const BASELINE: &str = "baseline.json";
const SCOPE: &str = "scope.json";
const SCOPE_INTENT: &str = "scope-intent.json";
const INCARNATION: &str = "incarnation.json";
const INTENT: &str = "intent.json";
const FOREIGN_QUEUE: &str = "foreign-queue.json";
const CONFLICTS: &str = "conflicts.jsonl";
const CONFLICTS_PREV: &str = "conflicts.1.jsonl";
const CONFLICTS_DROPPED: &str = "conflicts.dropped";
const CONFLICTS_MAX_BYTES: u64 = 1 << 20;
fn write_atomic(path: &Path, bytes: &[u8]) -> LeanResult<()> {
super::safefs::check_parent(path)?;
let tmp = path.with_extension("tmp");
super::safefs::write_via_tmp(path, &tmp, bytes, None).map(|_| ())
}
impl SyncerState {
pub fn open(dir: PathBuf) -> LeanResult<SyncerState> {
fs::create_dir_all(&dir)?;
let lock = fs::OpenOptions::new()
.create(true)
.truncate(false)
.write(true)
.open(dir.join(LOCK))?;
#[cfg(unix)]
{
use std::os::fd::AsRawFd;
let rc = unsafe { libc::flock(lock.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
if rc != 0 {
return Err(LeanError::State(format!(
"another flint-sync already holds this workspace ({}): \
refusing to run two syncers over one tree",
dir.display()
)));
}
}
Ok(SyncerState { dir, _lock: lock })
}
pub fn marker_present(&self) -> bool {
self.dir.join(MARKER).exists()
}
pub fn sync_tree(&self) -> LeanResult<()> {
super::safefs::sync_tree(&self.dir)
}
pub fn write_drained(&self, seq: Option<u64>, acks: usize) -> LeanResult<()> {
let doc = serde_json::json!({
"unix": super::now_unix(),
"seq": seq,
"acks": acks,
});
write_atomic(&self.dir.join(DRAINED), doc.to_string().as_bytes())
}
pub fn clear_drained(&self) -> LeanResult<()> {
match fs::remove_file(self.dir.join(DRAINED)) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(LeanError::State(format!("clear {DRAINED}: {e}"))),
}
}
pub fn drained_path(&self) -> PathBuf {
self.dir.join(DRAINED)
}
pub fn write_marker(&self) -> LeanResult<()> {
write_atomic(&self.dir.join(MARKER), b"ok\n")
}
pub fn load_baseline(&self) -> LeanResult<Baseline> {
let p = self.dir.join(BASELINE);
if !p.exists() {
return Ok(Baseline::default());
}
let bytes = fs::read(&p)?;
serde_json::from_slice(&bytes).map_err(|e| LeanError::State(format!("baseline: {e}")))
}
pub fn save_baseline(&self, b: &Baseline) -> LeanResult<()> {
let bytes =
serde_json::to_vec(b).map_err(|e| LeanError::State(format!("baseline: {e}")))?;
write_atomic(&self.dir.join(BASELINE), &bytes)
}
pub fn load_scope(&self) -> LeanResult<Option<Vec<String>>> {
let p = self.dir.join(SCOPE);
let bytes = match fs::read(&p) {
Ok(b) => b,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(e) => {
return Err(LeanError::State(format!(
"cannot read the checkout scope at {}: {e} — refusing to \
guess what this workspace is holding",
p.display()
)))
}
};
serde_json::from_slice(&bytes).map_err(|e| LeanError::State(format!("scope: {e}")))
}
pub fn save_scope(&self, scope: Option<&[String]>) -> LeanResult<()> {
let p = self.dir.join(SCOPE);
match scope {
Some(entries) => {
let bytes = serde_json::to_vec(entries)
.map_err(|e| LeanError::State(format!("scope: {e}")))?;
write_atomic(&p, &bytes)
}
None => match fs::remove_file(&p) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(LeanError::State(format!("clear {SCOPE}: {e}"))),
},
}
}
pub fn load_scope_intent(&self) -> LeanResult<Option<ScopeIntent>> {
let p = self.dir.join(SCOPE_INTENT);
let bytes = match fs::read(&p) {
Ok(b) => b,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(e) => {
return Err(LeanError::State(format!(
"cannot read the rescope intent at {}: {e} — refusing to run a barrier \
over a half-applied scope",
p.display()
)))
}
};
serde_json::from_slice(&bytes)
.map(Some)
.map_err(|e| LeanError::State(format!("scope intent: {e}")))
}
pub fn save_scope_intent(&self, intent: &ScopeIntent) -> LeanResult<()> {
let bytes = serde_json::to_vec(intent)
.map_err(|e| LeanError::State(format!("scope intent: {e}")))?;
write_atomic(&self.dir.join(SCOPE_INTENT), &bytes)
}
pub fn clear_scope_intent(&self) -> LeanResult<()> {
match fs::remove_file(self.dir.join(SCOPE_INTENT)) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(LeanError::State(format!("clear {SCOPE_INTENT}: {e}"))),
}
}
pub fn load_incarnation(&self) -> LeanResult<Option<Incarnation>> {
let p = self.dir.join(INCARNATION);
if !p.exists() {
return Ok(None);
}
let bytes = fs::read(&p)?;
serde_json::from_slice(&bytes)
.map(Some)
.map_err(|e| LeanError::State(format!("incarnation: {e}")))
}
pub fn save_incarnation(&self, i: &Incarnation) -> LeanResult<()> {
let bytes = serde_json::to_vec_pretty(i)
.map_err(|e| LeanError::State(format!("incarnation: {e}")))?;
write_atomic(&self.dir.join(INCARNATION), &bytes)
}
pub fn load_foreign_queue(&self) -> LeanResult<Vec<ForeignChange>> {
match fs::read(self.dir.join(FOREIGN_QUEUE)) {
Ok(bytes) => serde_json::from_slice(&bytes)
.map_err(|e| LeanError::State(format!("foreign queue: {e}"))),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(vec![]),
Err(e) => Err(e.into()),
}
}
pub fn save_foreign_queue(&self, q: &[ForeignChange]) -> LeanResult<()> {
let bytes =
serde_json::to_vec(q).map_err(|e| LeanError::State(format!("foreign queue: {e}")))?;
write_atomic(&self.dir.join(FOREIGN_QUEUE), &bytes)
}
pub fn queue_foreign(&self, changes: &[ForeignChange]) -> LeanResult<()> {
let mut q = self.load_foreign_queue()?;
for c in changes {
q.retain(|x| x.path != c.path);
q.push(c.clone());
}
self.save_foreign_queue(&q)
}
pub fn requeue_installed_foreign(&self) -> LeanResult<usize> {
let mut j = self.load_intent()?;
if j.installed_foreign.is_empty() {
return Ok(0);
}
let n = j.installed_foreign.len();
self.queue_foreign(&j.installed_foreign)?;
j.installed_foreign.clear();
self.save_intent(&j)?;
Ok(n)
}
pub fn load_intent(&self) -> LeanResult<IntentJournal> {
let p = self.dir.join(INTENT);
if !p.exists() {
return Ok(IntentJournal::default());
}
let bytes = fs::read(&p)?;
serde_json::from_slice(&bytes).map_err(|e| LeanError::State(format!("intent: {e}")))
}
pub fn save_intent(&self, j: &IntentJournal) -> LeanResult<()> {
let bytes =
serde_json::to_vec_pretty(j).map_err(|e| LeanError::State(format!("intent: {e}")))?;
write_atomic(&self.dir.join(INTENT), &bytes)
}
pub fn clear_intent_keys(&self) -> LeanResult<()> {
let mut j = self.load_intent()?;
if !j.flush_uuid.is_empty() {
if !j.recent_uuids.contains(&j.flush_uuid) {
j.recent_uuids.push(j.flush_uuid.clone());
}
let excess = j.recent_uuids.len().saturating_sub(8);
if excess > 0 {
j.recent_uuids.drain(..excess);
}
}
j.flush_uuid = String::new();
j.keys.clear();
j.declared_deletes.clear();
j.installed_foreign.clear();
self.save_intent(&j)
}
pub fn append_conflict(&self, c: &ConflictRecord) -> LeanResult<()> {
use std::io::Write;
let line =
serde_json::to_string(c).map_err(|e| LeanError::State(format!("conflict: {e}")))?;
let live = self.dir.join(CONFLICTS);
let mut f = fs::OpenOptions::new().create(true).append(true).open(&live)?;
writeln!(f, "{line}")?;
let len = f.metadata()?.len();
drop(f);
if len > CONFLICTS_MAX_BYTES {
self.rotate_conflicts()?;
}
Ok(())
}
fn rotate_conflicts(&self) -> LeanResult<()> {
let live = self.dir.join(CONFLICTS);
let prev = self.dir.join(CONFLICTS_PREV);
if let Ok(text) = fs::read_to_string(&prev) {
let dropped = text.lines().filter(|l| !l.trim().is_empty()).count() as u64;
let total = self.conflicts_dropped()? + dropped;
write_atomic(&self.dir.join(CONFLICTS_DROPPED), total.to_string().as_bytes())?;
}
fs::rename(&live, &prev)?;
Ok(())
}
pub fn conflicts_dropped(&self) -> LeanResult<u64> {
match fs::read_to_string(self.dir.join(CONFLICTS_DROPPED)) {
Ok(s) => Ok(s.trim().parse().unwrap_or(0)),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(0),
Err(e) => Err(LeanError::State(format!("{CONFLICTS_DROPPED}: {e}"))),
}
}
pub fn load_conflicts(&self) -> LeanResult<Vec<ConflictRecord>> {
let mut out = vec![];
for name in [CONFLICTS_PREV, CONFLICTS] {
let p = self.dir.join(name);
let text = match fs::read_to_string(&p) {
Ok(t) => t,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
Err(e) => {
return Err(LeanError::State(format!(
"cannot read {name}: {e} — refusing to report an unreadable \
conflict log as an empty one"
)))
}
};
for line in text.lines() {
if line.trim().is_empty() {
continue;
}
out.push(
serde_json::from_str(line)
.map_err(|e| LeanError::State(format!("conflict line: {e}")))?,
);
}
}
Ok(out)
}
}