use std::collections::{BTreeMap, BTreeSet};
use serde::{Deserialize, Serialize};
use flint_store::{
crc64_nvme, GenerationStamps, ObjectMeta, ObjectStore, PutCondition, StoreError,
};
use super::{LeanConfig, LeanError, LeanResult};
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct LeanEntry {
pub key: String,
pub etag: String,
pub crc64_b64: String,
pub size: u64,
pub mode: u32,
pub mtime_unix: i64,
pub generation: u64,
pub epoch: u64,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct LeanManifest {
pub seq: u64,
pub entries: BTreeMap<String, LeanEntry>,
#[serde(default)]
pub sole_writer: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub boundary_source: Option<String>,
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub tombstones: BTreeMap<String, Tombstone>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct Tombstone {
pub etag: String,
pub seq: u64,
}
pub const TOMBSTONE_KEEP_SEQS: u64 = 10_000;
impl LeanManifest {
pub fn to_bytes(&self) -> Vec<u8> {
serde_json::to_vec(self).expect("manifest serializes")
}
pub fn parse(bytes: &[u8]) -> Result<LeanManifest, String> {
serde_json::from_slice(bytes).map_err(|e| e.to_string())
}
}
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct Pointer {
pub seq: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub entries_key: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub entries_seq: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub chunks: Option<Vec<super::chunk::ChunkRef>>,
#[serde(default)]
pub sole_writer: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub boundary_source: Option<String>,
pub epoch: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tombstones: Option<String>,
}
#[derive(Debug)]
pub enum Entries<'a> {
Single { key: &'a str, seq: u64 },
Chunked(&'a [super::chunk::ChunkRef]),
}
impl Pointer {
pub fn entries(&self) -> LeanResult<Entries<'_>> {
match (&self.entries_key, &self.chunks) {
(Some(k), None) => {
Ok(Entries::Single { key: k, seq: self.entries_seq.unwrap_or(self.seq) })
}
(None, Some(c)) => Ok(Entries::Chunked(c)),
(Some(_), Some(_)) => Err(super::LeanError::State(format!(
"manifest pointer at seq {} names BOTH a generation object and a chunk list; \
refusing to guess which describes this manifest",
self.seq
))),
(None, None) => Err(super::LeanError::State(format!(
"manifest pointer at seq {} names no entries at all — neither a generation \
object nor a chunk list; an empty project is an empty chunk LIST, not a \
missing one",
self.seq
))),
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct ManifestHandle {
pub etag: String,
pub legacy: bool,
pub prev_chunks: Vec<super::chunk::ChunkRef>,
pub prev_tombstones: Option<String>,
}
pub const REFUSAL_DOC: &str = concat!(
"{\"moved\":\".flint/lean/current\",",
"\"note\":\"this workspace uses the manifest pointer layout; upgrade flint-sync\"}"
);
pub struct LoadedManifest {
pub manifest: LeanManifest,
pub etag: String,
pub last_modified_unix: Option<u64>,
pub pointer: Option<Pointer>,
}
impl LoadedManifest {
pub fn handle(&self) -> ManifestHandle {
ManifestHandle {
etag: self.etag.clone(),
legacy: self.pointer.is_none(),
prev_chunks: match self.pointer.as_ref().map(|p| p.entries()) {
Some(Ok(Entries::Chunked(c))) => c.to_vec(),
_ => Vec::new(),
},
prev_tombstones: self.pointer.as_ref().and_then(|p| p.tombstones.clone()),
}
}
}
pub struct LoadedPointer {
pub pointer: Pointer,
pub etag: String,
pub last_modified_unix: Option<u64>,
}
pub async fn load_pointer(
store: &dyn ObjectStore,
cfg: &LeanConfig,
) -> LeanResult<Option<LoadedPointer>> {
match store.get_whole(&cfg.current_key(), None).await {
Ok((meta, bytes)) => {
let pointer: Pointer = serde_json::from_slice(&bytes)
.map_err(|e| super::LeanError::State(format!("manifest pointer parse: {e}")))?;
Ok(Some(LoadedPointer { pointer, etag: meta.etag, last_modified_unix: meta.last_modified_unix }))
}
Err(StoreError::NotFound(_)) => Ok(None),
Err(e) => Err(e.into()),
}
}
const LOAD_ATTEMPTS: u32 = 3;
pub async fn load(
store: &dyn ObjectStore,
cfg: &LeanConfig,
) -> LeanResult<Option<LoadedManifest>> {
load_from(store, cfg, None).await
}
pub async fn load_from(
store: &dyn ObjectStore,
cfg: &LeanConfig,
mut first: Option<Option<LoadedPointer>>,
) -> LeanResult<Option<LoadedManifest>> {
for attempt in 0..LOAD_ATTEMPTS {
let read = match first.take() {
Some(already) => already,
None => load_pointer(store, cfg).await?,
};
if let Some(LoadedPointer { pointer: p, etag, last_modified_unix: pointer_lm }) = read {
let (mut manifest, meta_lm) = match p.entries()? {
Entries::Single { key, .. } => {
let (meta, bytes) = match store.get_whole(key, None).await {
Ok(v) => v,
Err(StoreError::NotFound(_)) => {
return Err(super::LeanError::State(format!(
"manifest pointer at seq {} names {key}, which does not exist — \
refusing to treat a broken pointer as an empty workspace",
p.seq
)))
}
Err(e) => return Err(e.into()),
};
let m = LeanManifest::parse(&bytes)
.map_err(|e| super::LeanError::State(format!("manifest parse: {e}")))?;
(m, meta.last_modified_unix)
}
Entries::Chunked(refs) => {
let mut bodies = Vec::with_capacity(refs.len());
let mut raced = false;
for r in refs {
let key = cfg.chunk_key(&r.addr);
match store.get_whole(&key, None).await {
Ok((_, b)) => bodies.push(b.to_vec()),
Err(StoreError::NotFound(_)) => {
let moved = match load_pointer(store, cfg).await? {
Some(now) => now.etag != etag,
None => true,
};
if moved && attempt + 1 < LOAD_ATTEMPTS {
raced = true;
break;
}
return Err(super::LeanError::State(format!(
"manifest pointer at seq {} names chunk {} (first key {:?}), \
which does not exist — refusing to serve a manifest with a \
hole in it as though those entries had been deleted",
p.seq, r.addr, r.first
)));
}
Err(e) => return Err(e.into()),
}
}
if raced {
continue;
}
let entries = super::chunk::assemble(refs, &bodies)?;
let tombstones = match &p.tombstones {
None => BTreeMap::new(),
Some(addr) => match store.get_whole(&cfg.chunk_key(addr), None).await {
Ok((_, b)) => super::chunk::decode_tombstones(&b, addr)?,
Err(StoreError::NotFound(_)) => {
let moved = match load_pointer(store, cfg).await? {
Some(now) => now.etag != etag,
None => true,
};
if moved && attempt + 1 < LOAD_ATTEMPTS {
continue;
}
return Err(super::LeanError::State(format!(
"manifest pointer at seq {} names tombstones {addr}, which do not exist",
p.seq
)));
}
Err(e) => return Err(e.into()),
},
};
(
LeanManifest {
seq: p.seq,
entries,
sole_writer: p.sole_writer,
boundary_source: p.boundary_source.clone(),
tombstones,
},
None,
)
}
};
manifest.seq = p.seq;
manifest.sole_writer = p.sole_writer;
manifest.boundary_source = p.boundary_source.clone();
return Ok(Some(LoadedManifest {
manifest,
etag,
last_modified_unix: pointer_lm.or(meta_lm),
pointer: Some(p),
}));
}
return match store.get_whole(&cfg.manifest_key(), None).await {
Ok((meta, bytes)) => {
let manifest = LeanManifest::parse(&bytes)
.map_err(|e| super::LeanError::State(format!("manifest parse: {e}")))?;
Ok(Some(LoadedManifest {
manifest,
etag: meta.etag,
last_modified_unix: meta.last_modified_unix,
pointer: None,
}))
}
Err(StoreError::NotFound(_)) => Ok(None),
Err(e) => Err(e.into()),
};
}
Err(super::LeanError::State(format!(
"manifest load restarted {LOAD_ATTEMPTS} times without settling — a sweep is \
collecting generations faster than this reader can finish one"
)))
}
pub const EDIT_CAS_ATTEMPTS: usize = 16;
#[derive(Debug)]
pub enum EditError<E> {
Refused(E),
Store(LeanError),
Contended,
}
pub async fn commit_edit<E>(
store: &dyn ObjectStore,
cfg: &LeanConfig,
flush: &str,
edit: impl Fn(Option<&LoadedManifest>, &mut LeanManifest) -> Result<(), E>,
) -> Result<(), EditError<E>> {
for _ in 0..EDIT_CAS_ATTEMPTS {
let current = load(store, cfg).await.map_err(EditError::Store)?;
let mut doc = current.as_ref().map(|l| l.manifest.clone()).unwrap_or_default();
doc.seq += 1;
doc.sole_writer = false;
doc.boundary_source = None;
edit(current.as_ref(), &mut doc).map_err(EditError::Refused)?;
let seq = doc.seq;
doc.tombstones.retain(|_, t| seq.saturating_sub(t.seq) <= TOMBSTONE_KEEP_SEQS);
let expected = current.as_ref().map(|l| l.handle());
let before = current.as_ref().map(|l| &l.manifest);
match cas_write_retiring(store, cfg, &doc, before, expected.as_ref(), 0, flush, None).await {
Ok(_) => return Ok(()),
Err(LeanError::Store(StoreError::PreconditionFailed(_))) | Err(LeanError::Store(StoreError::Conflict(_))) => {
continue
}
Err(e) => return Err(EditError::Store(e)),
}
}
Err(EditError::Contended)
}
pub async fn cas_write(
store: &dyn ObjectStore,
cfg: &LeanConfig,
m: &LeanManifest,
expected: Option<&ManifestHandle>,
epoch: u64,
flush_uuid: &str,
) -> LeanResult<ObjectMeta> {
cas_write_stamped(store, cfg, m, expected, epoch, flush_uuid, None).await
}
pub async fn cas_write_stamped(
store: &dyn ObjectStore,
cfg: &LeanConfig,
m: &LeanManifest,
expected: Option<&ManifestHandle>,
epoch: u64,
flush_uuid: &str,
boundary_source: Option<&str>,
) -> LeanResult<ObjectMeta> {
Ok(cas_write_refs(store, cfg, m, expected, epoch, flush_uuid, boundary_source).await?.0)
}
async fn cas_write_refs(
store: &dyn ObjectStore,
cfg: &LeanConfig,
m: &LeanManifest,
expected: Option<&ManifestHandle>,
epoch: u64,
flush_uuid: &str,
boundary_source: Option<&str>,
) -> LeanResult<(ObjectMeta, Option<Vec<String>>)> {
if cfg.chunked {
let (meta, refs) = cas_write_chunked_refs(
store,
cfg,
m,
expected,
expected.map(|h| h.prev_chunks.as_slice()).unwrap_or(&[]),
PublishStamps { epoch, flush_uuid, boundary_source },
)
.await?;
return Ok((meta, Some(refs)));
}
let mut m = m.clone();
if boundary_source.is_some() {
m.boundary_source = boundary_source.map(|s| s.to_string());
}
let m = &m;
let bytes = m.to_bytes();
let crc = crc64_nvme(&bytes);
let stamps = GenerationStamps {
generation: m.seq,
epoch,
flush_uuid: flush_uuid.to_string(),
boundary_source: m.boundary_source.clone(),
posix: None,
};
let entries_key = cfg.generation_key(m.seq, flush_uuid);
match store
.put_whole(&entries_key, bytes.into(), &PutCondition::IfNoneMatchAny, &stamps, crc)
.await
{
Ok(_) => {}
Err(StoreError::PreconditionFailed(_)) | Err(StoreError::Conflict(_)) => {}
Err(e) => return Err(e.into()),
}
let pointer = Pointer {
seq: m.seq,
entries_key: Some(entries_key.clone()),
entries_seq: Some(m.seq),
chunks: None,
sole_writer: m.sole_writer,
boundary_source: m.boundary_source.clone(),
epoch,
tombstones: None,
};
Ok((put_pointer(store, cfg, &pointer, expected, &stamps).await?, None))
}
pub struct PublishStamps<'a> {
pub epoch: u64,
pub flush_uuid: &'a str,
pub boundary_source: Option<&'a str>,
}
pub async fn cas_write_chunked(
store: &dyn ObjectStore,
cfg: &LeanConfig,
m: &LeanManifest,
expected: Option<&ManifestHandle>,
prev: &[super::chunk::ChunkRef],
pub_stamps: PublishStamps<'_>,
) -> LeanResult<ObjectMeta> {
Ok(cas_write_chunked_refs(store, cfg, m, expected, prev, pub_stamps).await?.0)
}
async fn cas_write_chunked_refs(
store: &dyn ObjectStore,
cfg: &LeanConfig,
m: &LeanManifest,
expected: Option<&ManifestHandle>,
prev: &[super::chunk::ChunkRef],
pub_stamps: PublishStamps<'_>,
) -> LeanResult<(ObjectMeta, Vec<String>)> {
let PublishStamps { epoch, flush_uuid, boundary_source } = pub_stamps;
let mut m = m.clone();
if boundary_source.is_some() {
m.boundary_source = boundary_source.map(|s| s.to_string());
}
let m = &m;
let split =
super::chunk::split_with(&m.entries, cfg.chunk_target, cfg.chunk_min, cfg.chunk_max)?;
let have: std::collections::HashSet<&str> = prev.iter().map(|r| r.addr.as_str()).collect();
let stamps = GenerationStamps {
generation: m.seq,
epoch,
flush_uuid: flush_uuid.to_string(),
boundary_source: m.boundary_source.clone(),
posix: None,
};
for (r, body) in &split {
if have.contains(r.addr.as_str()) {
continue;
}
let key = cfg.chunk_key(&r.addr);
let crc = crc64_nvme(body);
store.put_whole(&key, body.clone().into(), &PutCondition::Unconditional, &stamps, crc).await?;
}
let tombstones = if m.tombstones.is_empty() {
None
} else {
let (addr, body) = super::chunk::encode_tombstones(&m.tombstones)?;
let crc = crc64_nvme(&body);
store.put_whole(&cfg.chunk_key(&addr), body.into(), &PutCondition::Unconditional, &stamps, crc).await?;
Some(addr)
};
let pointer = Pointer {
seq: m.seq,
entries_key: None,
entries_seq: None,
chunks: Some(split.into_iter().map(|(r, _)| r).collect()),
sole_writer: m.sole_writer,
boundary_source: m.boundary_source.clone(),
epoch,
tombstones: tombstones.clone(),
};
let meta = put_pointer(store, cfg, &pointer, expected, &stamps).await?;
let mut refs: Vec<String> = pointer.chunks.iter().flatten().map(|r| r.addr.clone()).collect();
refs.extend(tombstones);
Ok((meta, refs))
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct RetireLog {
pub at_unix: u64,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub handles: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub chunks: Vec<String>,
}
pub fn retired_prefix(cfg: &LeanConfig) -> String {
format!("{}/{}/retired/", cfg.prefix, super::LEAN_DIR)
}
pub fn retire_log_at(cfg: &LeanConfig, key: &str) -> Option<u64> {
key.strip_prefix(&retired_prefix(cfg))?.get(..20)?.parse().ok()
}
#[allow(clippy::too_many_arguments)]
pub async fn cas_write_retiring(
store: &dyn ObjectStore,
cfg: &LeanConfig,
m: &LeanManifest,
before: Option<&LeanManifest>,
expected: Option<&ManifestHandle>,
epoch: u64,
flush_uuid: &str,
boundary_source: Option<&str>,
) -> LeanResult<ObjectMeta> {
let (meta, refs) = cas_write_refs(store, cfg, m, expected, epoch, flush_uuid, boundary_source).await?;
let still: std::collections::HashSet<&str> = m.entries.values().map(|e| e.key.as_str()).collect();
let mut handles: Vec<String> = before
.map(|b| b.entries.values().map(|e| e.key.clone()).filter(|k| !still.contains(k.as_str())).collect())
.unwrap_or_default();
handles.sort();
handles.dedup();
let mut chunks: Vec<String> = vec![];
if let (Some(refs), Some(h)) = (refs.as_ref(), expected) {
let now: std::collections::HashSet<&str> = refs.iter().map(|a| a.as_str()).collect();
chunks = h
.prev_chunks
.iter()
.map(|r| r.addr.clone())
.chain(h.prev_tombstones.clone())
.filter(|a| !now.contains(a.as_str()))
.collect();
chunks.sort();
chunks.dedup();
}
if !handles.is_empty() || !chunks.is_empty() {
let at = super::now_unix();
let log = RetireLog { at_unix: at, handles, chunks };
let key = format!("{}{at:020}-{flush_uuid}-{}.json", retired_prefix(cfg), uuid::Uuid::new_v4());
let body = serde_json::to_vec(&log).map_err(|e| LeanError::State(format!("retire log: {e}")))?;
let crc = crc64_nvme(&body);
let stamps = GenerationStamps {
generation: m.seq,
epoch,
flush_uuid: flush_uuid.to_string(),
boundary_source: None,
posix: None,
};
if let Err(e) = store.put_whole(&key, body.into(), &PutCondition::IfNoneMatchAny, &stamps, crc).await {
eprintln!("flint-sync: could not log what seq {} retired ({e}); the sweeps judge it by write age", m.seq);
}
}
Ok(meta)
}
async fn put_pointer(
store: &dyn ObjectStore,
cfg: &LeanConfig,
pointer: &Pointer,
expected: Option<&ManifestHandle>,
stamps: &GenerationStamps,
) -> LeanResult<ObjectMeta> {
let body = serde_json::to_vec(pointer)
.map_err(|e| super::LeanError::State(format!("pointer encode: {e}")))?;
let crc = crc64_nvme(&body);
let cond = match expected {
Some(h) if !h.legacy => PutCondition::IfMatch(h.etag.clone()),
_ => PutCondition::IfNoneMatchAny,
};
let meta = store.put_whole(&cfg.current_key(), body.into(), &cond, stamps, crc).await?;
if let Some(h) = expected.filter(|h| h.legacy) {
let bytes = REFUSAL_DOC.as_bytes().to_vec();
let crc = crc64_nvme(&bytes);
if let Err(e) = store
.put_whole(
&cfg.manifest_key(),
bytes.into(),
&PutCondition::IfMatch(h.etag.clone()),
stamps,
crc,
)
.await
{
eprintln!("flint-sync: could not poison the legacy manifest key after migration: {e}");
}
}
Ok(meta)
}
pub async fn rotate_for_takeover(
store: &dyn ObjectStore,
cfg: &LeanConfig,
epoch: u64,
) -> LeanResult<Option<(LeanManifest, String)>> {
for _ in 0..3 {
let Some(LoadedPointer { pointer: p, etag, .. }) = load_pointer(store, cfg).await? else {
let Some(loaded) = load(store, cfg).await? else {
return Ok(None);
};
let mut rotated = loaded.manifest.clone();
rotated.seq += 1;
let h = loaded.handle();
match cas_write(store, cfg, &rotated, Some(&h), epoch, "takeover-rotation").await {
Ok(meta) => return Ok(Some((rotated, meta.etag))),
Err(super::LeanError::Store(StoreError::PreconditionFailed(_) | StoreError::Conflict(_))) => continue,
Err(e) => return Err(e),
}
};
let next = Pointer { seq: p.seq + 1, epoch, ..p.clone() };
let stamps = GenerationStamps {
generation: next.seq,
epoch,
flush_uuid: "takeover-rotation".to_string(),
boundary_source: next.boundary_source.clone(),
posix: None,
};
let h = ManifestHandle { etag, legacy: false, prev_chunks: Vec::new(), prev_tombstones: None };
match put_pointer(store, cfg, &next, Some(&h), &stamps).await {
Ok(meta) => {
let mut m = LeanManifest::default();
m.seq = next.seq;
m.sole_writer = next.sole_writer;
m.boundary_source = next.boundary_source.clone();
return Ok(Some((m, meta.etag)));
}
Err(super::LeanError::Store(StoreError::PreconditionFailed(_) | StoreError::Conflict(_))) => continue,
Err(e) => return Err(e),
}
}
Err(super::LeanError::State(
"takeover rotation lost 3 CAS races — a live writer is still publishing".into(),
))
}
pub const KEEP_GENERATIONS: usize = 5;
pub const ORPHAN_GRACE_SECS: u64 = 3600;
pub const RETIRE_GRACE_SECS: u64 = 600;
pub async fn sweep_generations(store: &dyn ObjectStore, cfg: &LeanConfig) -> LeanResult<usize> {
let Some(lp) = load_pointer(store, cfg).await? else {
return Ok(0);
};
let Some(live) = lp.pointer.entries_key else {
let prefix = format!("{}/{}/manifests/", cfg.prefix, super::LEAN_DIR);
let now = super::now_unix();
let mut removed = 0;
for o in store.list(&prefix).await? {
let age = o.last_modified_unix.map(|t| now.saturating_sub(t));
if !age.is_some_and(|a| a >= cfg.orphan_grace_secs) {
continue;
}
match store.delete(&o.key).await {
Ok(()) => removed += 1,
Err(e) => eprintln!("flint-sync: could not reap migrated generation {}: {e}", o.key),
}
}
return Ok(removed);
};
let prefix = format!("{}/{}/manifests/", cfg.prefix, super::LEAN_DIR);
let mut listed = store.list(&prefix).await?;
listed.sort_by(|a, b| a.key.cmp(&b.key));
let now = super::now_unix();
let live_ix = listed.iter().position(|o| o.key == live);
let mut doomed: Vec<String> = Vec::new();
for (i, o) in listed.iter().enumerate() {
if o.key == live {
continue;
}
match live_ix {
Some(ix) if i > ix => {
let age = o.last_modified_unix.map(|t| now.saturating_sub(t));
if age.is_some_and(|a| a > ORPHAN_GRACE_SECS) {
doomed.push(o.key.clone());
}
}
Some(ix) => {
let superseded_for = listed[i + 1].last_modified_unix.map(|t| now.saturating_sub(t));
if ix.saturating_sub(i) > KEEP_GENERATIONS
&& superseded_for.is_some_and(|a| a >= cfg.retire_grace_secs)
{
doomed.push(o.key.clone());
}
}
None => return Ok(0),
}
}
let mut removed = 0;
for k in doomed {
match store.delete(&k).await {
Ok(()) => removed += 1,
Err(e) => eprintln!("flint-sync: could not reap superseded generation {k}: {e}"),
}
}
Ok(removed)
}
pub async fn sweep_chunks(store: &dyn ObjectStore, cfg: &LeanConfig) -> LeanResult<usize> {
sweep_chunks_sparing(store, cfg, &std::collections::HashSet::new()).await
}
pub async fn sweep_chunks_sparing(
store: &dyn ObjectStore,
cfg: &LeanConfig,
spared: &std::collections::HashSet<String>,
) -> LeanResult<usize> {
let Some(before) = load_pointer(store, cfg).await? else {
return Ok(0);
};
let mut refs: std::collections::HashSet<String> = match before.pointer.entries()? {
Entries::Chunked(c) => c.iter().map(|r| r.addr.clone()).collect(),
Entries::Single { .. } => return Ok(0),
};
if let Some(t) = &before.pointer.tombstones {
refs.insert(t.clone());
}
let prefix = format!("{}/{}/chunks/", cfg.prefix, super::LEAN_DIR);
let listed = store.list(&prefix).await?;
let after = load_pointer(store, cfg).await?;
if after.map(|p| p.etag) != Some(before.etag) {
return Ok(0);
}
let now = super::now_unix();
let mut removed = 0;
for o in listed {
let Some(addr) = o.key.rsplit('/').next() else { continue };
if refs.contains(addr) || spared.contains(&o.key) {
continue;
}
if let Some(listed) = o.last_modified_unix {
if now.saturating_sub(listed) < cfg.orphan_grace_secs {
continue;
}
}
let fresh = match store.head(&o.key).await {
Ok(m) => m.last_modified_unix,
Err(StoreError::NotFound(_)) => continue,
Err(e) => {
eprintln!("flint-sync: could not date orphan chunk {} ({e}); leaving it", o.key);
continue;
}
};
let age = fresh.map(|t| now.saturating_sub(t));
if !age.is_some_and(|a| a >= cfg.orphan_grace_secs) {
continue;
}
match store.delete(&o.key).await {
Ok(()) => removed += 1,
Err(e) => eprintln!("flint-sync: could not reap orphan chunk {}: {e}", o.key),
}
}
Ok(removed)
}
pub fn merge(
base: &BTreeMap<String, String>,
theirs: &LeanManifest,
mine_upserts: &BTreeMap<String, LeanEntry>,
mine_deletes: &BTreeSet<String>,
parked: &BTreeSet<String>,
) -> Merged {
let mut merged = theirs.clone();
merged.seq = theirs.seq + 1;
merged.sole_writer = false;
merged.boundary_source = None;
let mut foreign: Vec<(String, LeanEntry)> = vec![];
let mut overridden: Vec<(String, LeanEntry)> = vec![];
let recreated: Vec<(String, String)> = mine_upserts
.keys()
.filter(|p| !theirs.entries.contains_key(*p))
.filter_map(|p| {
let base_etag = base.get(p)?;
let deleted = theirs.tombstones.get(p).map(|t| t.etag.clone()).unwrap_or_else(|| base_etag.clone());
Some((p.clone(), deleted))
})
.collect();
for (p, e) in &theirs.entries {
let changed = base.get(p).map(|b| b != &e.etag).unwrap_or(true);
if changed && !mine_upserts.contains_key(p) && !parked.contains(p) {
foreign.push((p.clone(), e.clone()));
}
if changed && mine_upserts.contains_key(p) {
overridden.push((p.clone(), e.clone()));
}
}
for (p, e) in mine_upserts {
merged.entries.insert(p.clone(), e.clone());
merged.tombstones.remove(p);
}
let mut deletes = BTreeMap::new();
let mut over_theirs: Vec<(String, LeanEntry)> = vec![];
for p in mine_deletes {
let outcome = delete_outcome(base, theirs, p, parked);
deletes.insert(p.clone(), outcome);
if outcome == DeleteOutcome::OverTheirs {
if let Some(e) = theirs.entries.get(p) {
over_theirs.push((p.clone(), e.clone()));
}
}
if outcome != DeleteOutcome::Parked {
if let Some(retired) = merged.entries.remove(p) {
merged.tombstones.insert(p.clone(), Tombstone { etag: retired.etag, seq: merged.seq });
}
}
}
merged.tombstones.retain(|_, t| merged.seq.saturating_sub(t.seq) <= TOMBSTONE_KEEP_SEQS);
Merged { doc: merged, foreign, overridden, deletes, recreated, over_theirs }
}
pub struct Merged {
pub doc: LeanManifest,
pub foreign: Vec<(String, LeanEntry)>,
pub overridden: Vec<(String, LeanEntry)>,
pub deletes: BTreeMap<String, DeleteOutcome>,
pub recreated: Vec<(String, String)>,
pub over_theirs: Vec<(String, LeanEntry)>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeleteOutcome {
Applied,
OverTheirs,
Parked,
}
pub fn delete_outcome(
base: &BTreeMap<String, String>,
theirs: &LeanManifest,
p: &str,
parked: &BTreeSet<String>,
) -> DeleteOutcome {
if parked.contains(p) {
return DeleteOutcome::Parked;
}
match (theirs.entries.get(p), base.get(p)) {
(Some(e), Some(b)) if &e.etag == b => DeleteOutcome::Applied,
(None, _) => DeleteOutcome::Applied,
(Some(_), _) => DeleteOutcome::OverTheirs,
}
}