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\"}"
);
#[derive(Clone)]
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()),
}
}
pub(crate) const LOAD_ATTEMPTS: u32 = 3;
enum Fetched {
Complete { bodies: Vec<bytes::Bytes>, tombstones: BTreeMap<String, Tombstone> },
Missing(Hole),
}
pub(crate) enum Hole {
Chunk { addr: String, first: String },
Tombstones(String),
}
impl Hole {
pub(crate) fn refusal(&self, seq: u64) -> String {
match self {
Hole::Chunk { addr, first } => format!(
"manifest pointer at seq {seq} names chunk {addr} (first key {first:?}), which does \
not exist — refusing to serve a manifest with a hole in it as though those entries \
had been deleted"
),
Hole::Tombstones(addr) => {
format!("manifest pointer at seq {seq} names tombstones {addr}, which do not exist")
}
}
}
}
async fn fetch_chunks(
store: &dyn ObjectStore,
cfg: &LeanConfig,
p: &Pointer,
refs: &[super::chunk::ChunkRef],
) -> LeanResult<Fetched> {
use futures::stream::{self, StreamExt};
let cache = cfg.chunk_cache.as_deref();
let mut bodies: Vec<Option<bytes::Bytes>> = vec![None; refs.len()];
let mut tombstones: Option<BTreeMap<String, Tombstone>> = None;
if let Some(c) = cache {
for (slot, r) in bodies.iter_mut().zip(refs) {
if let Some(b) = c.get(&r.addr) {
if super::chunk::chunk_address(&b) == r.addr {
*slot = Some(b);
} else {
c.remove(&r.addr);
}
}
}
if let Some(addr) = &p.tombstones {
if let Some(b) = c.get(addr) {
match super::chunk::decode_tombstones(&b, addr) {
Ok(t) => tombstones = Some(t),
Err(_) => c.remove(addr),
}
}
}
}
let cached: Vec<usize> = (0..refs.len()).filter(|&i| bodies[i].is_some()).collect();
let tombstones_cached = tombstones.is_some();
let fetch_round = |want_chunks: Vec<usize>, want_tombstones: bool| async move {
let mut wants: Vec<(Option<usize>, String)> =
want_chunks.into_iter().map(|i| (Some(i), refs[i].addr.clone())).collect();
if want_tombstones {
if let Some(a) = &p.tombstones {
wants.push((None, a.clone()));
}
}
stream::iter(wants.into_iter().map(|(slot, addr)| async move {
let got = store.get_whole(&cfg.chunk_key(&addr), None).await;
(slot, addr, got)
}))
.buffer_unordered(cfg.chunk_fetch_parallelism.max(1))
.collect::<Vec<_>>()
.await
};
let misses: Vec<usize> = (0..refs.len()).filter(|&i| bodies[i].is_none()).collect();
let tombstones_missing = p.tombstones.is_some() && !tombstones_cached;
let listing = async {
if cached.is_empty() && !tombstones_cached {
return None;
}
Some(store.list(&format!("{}/{}/chunks/", cfg.prefix, super::LEAN_DIR)).await)
};
let (listing, fetched) = futures::join!(listing, fetch_round(misses, tombstones_missing));
let mut rounds = vec![fetched];
match listing {
None => {}
Some(Ok(listed)) => {
let present: std::collections::HashSet<&str> =
listed.iter().filter_map(|o| o.key.rsplit('/').next()).collect();
for &i in &cached {
if !present.contains(refs[i].addr.as_str()) {
return Ok(Fetched::Missing(Hole::Chunk {
addr: refs[i].addr.clone(),
first: refs[i].first.clone(),
}));
}
}
if let (true, Some(addr)) = (tombstones_cached, &p.tombstones) {
if !present.contains(addr.as_str()) {
return Ok(Fetched::Missing(Hole::Tombstones(addr.clone())));
}
}
}
Some(Err(_)) => {
rounds.push(fetch_round(cached, tombstones_cached).await);
}
}
for (slot, addr, got) in rounds.into_iter().flatten() {
let body = match got {
Ok((_, b)) => b,
Err(StoreError::NotFound(_)) => {
return Ok(Fetched::Missing(match slot {
Some(i) => Hole::Chunk { addr, first: refs[i].first.clone() },
None => Hole::Tombstones(addr),
}));
}
Err(e) => return Err(e.into()),
};
match slot {
Some(i) => {
super::chunk::check_address_of(&body, &addr)?;
if let Some(c) = cache {
c.put(&addr, &body);
}
bodies[i] = Some(body);
}
None => {
tombstones = Some(super::chunk::decode_tombstones(&body, &addr)?);
if let Some(c) = cache {
c.put(&addr, &body);
}
}
}
}
Ok(Fetched::Complete {
bodies: bodies.into_iter().map(|b| b.expect("every chunk is cached or fetched")).collect(),
tombstones: tombstones.unwrap_or_default(),
})
}
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 (bodies, tombstones) = match fetch_chunks(store, cfg, &p, refs).await? {
Fetched::Complete { bodies, tombstones } => (bodies, tombstones),
Fetched::Missing(hole) => {
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(hole.refusal(p.seq)));
}
};
let entries = super::chunk::assemble_verified(refs, &bodies)?;
super::docview::remember(cfg, refs, &entries);
if let Some(c) = cfg.chunk_cache.as_deref() {
let mut live: std::collections::HashSet<String> = refs.iter().map(|r| r.addr.clone()).collect();
live.extend(p.tombstones.clone());
c.retain(&live);
}
(
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>> {
commit_edit_from(store, cfg, flush, None, edit).await
}
pub async fn commit_edit_from<E>(
store: &dyn ObjectStore,
cfg: &LeanConfig,
flush: &str,
first: Option<Option<std::sync::Arc<LoadedManifest>>>,
edit: impl Fn(Option<&LoadedManifest>, &mut LeanManifest) -> Result<(), E>,
) -> Result<(), EditError<E>> {
let mut first = first;
let mut sent: Option<LeanManifest> = None;
for _ in 0..EDIT_CAS_ATTEMPTS {
let current: Option<std::sync::Arc<LoadedManifest>> = match first.take() {
Some(read) => read,
None => load(store, cfg).await.map_err(EditError::Store)?.map(std::sync::Arc::new),
};
if let (Some(s), Some(c)) = (sent.as_ref(), current.as_ref()) {
if same_document(s, &c.manifest) {
return Ok(());
}
}
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_deref(), &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(_))) => {
sent = Some(doc);
continue;
}
Err(e) => return Err(EditError::Store(e)),
}
}
Err(EditError::Contended)
}
fn same_document(a: &LeanManifest, b: &LeanManifest) -> bool {
a.seq == b.seq
&& a.entries == b.entries
&& a.tombstones == b.tombstones
&& a.sole_writer == b.sole_writer
&& a.boundary_source == b.boundary_source
}
pub struct DocEdit<'a> {
pub seq: u64,
pub entries: EditEntries<'a>,
pub tombstones: BTreeMap<String, Tombstone>,
}
pub struct EditEntries<'a> {
base: &'a BTreeMap<String, LeanEntry>,
overlay: BTreeMap<String, Option<LeanEntry>>,
}
impl EditEntries<'_> {
pub fn get(&self, p: &str) -> Option<&LeanEntry> {
match self.overlay.get(p) {
Some(v) => v.as_ref(),
None => self.base.get(p),
}
}
pub fn contains_key(&self, p: &str) -> bool {
self.get(p).is_some()
}
pub fn insert(&mut self, p: String, e: LeanEntry) -> Option<LeanEntry> {
let was = self.get(&p).cloned();
self.overlay.insert(p, Some(e));
was
}
pub fn remove(&mut self, p: &str) -> Option<LeanEntry> {
let was = self.get(p).cloned()?;
self.overlay.insert(p.to_string(), None);
Some(was)
}
}
pub struct EditOutcome<E> {
pub result: Result<(), EditError<E>>,
pub held: Option<std::sync::Arc<LoadedManifest>>,
}
pub async fn commit_edit_paths<E>(
store: &dyn ObjectStore,
cfg: &LeanConfig,
flush: &str,
first: Option<Option<std::sync::Arc<LoadedManifest>>>,
edit: impl Fn(Option<&LoadedManifest>, &mut DocEdit<'_>) -> Result<(), E>,
) -> EditOutcome<E> {
use std::sync::Arc;
let mut first = first;
let mut sent: Option<Pointer> = None;
for _ in 0..EDIT_CAS_ATTEMPTS {
let current: Option<Arc<LoadedManifest>> = match first.take() {
Some(read) => read,
None => match load(store, cfg).await {
Ok(l) => l.map(Arc::new),
Err(e) => return EditOutcome { result: Err(EditError::Store(e)), held: None },
},
};
let chunked: Option<Vec<super::chunk::ChunkRef>> =
match current.as_ref().and_then(|l| l.pointer.as_ref()).map(|p| p.entries()) {
Some(Ok(Entries::Chunked(r))) if cfg.chunked && cfg.chunk_view => Some(r.to_vec()),
_ => None,
};
let refs = match chunked {
Some(r) => r,
None => {
let read = current.clone();
let result = commit_edit_from(store, cfg, flush, Some(current), |cur, doc| {
let base = doc.clone();
let mut de = DocEdit {
seq: doc.seq,
entries: EditEntries { base: &base.entries, overlay: BTreeMap::new() },
tombstones: std::mem::take(&mut doc.tombstones),
};
edit(cur, &mut de)?;
doc.tombstones = de.tombstones;
apply_overlay(&mut doc.entries, de.entries.overlay);
Ok(())
})
.await;
let held = match &result {
Err(EditError::Refused(_)) => read,
_ => None,
};
return EditOutcome { result, held };
}
};
let cur = current.as_deref().expect("chunked above");
if sent.as_ref().is_some_and(|s| cur.pointer.as_ref() == Some(s)) {
return EditOutcome { result: Ok(()), held: None };
}
let base = &cur.manifest;
let mut de = DocEdit {
seq: base.seq + 1,
entries: EditEntries { base: &base.entries, overlay: BTreeMap::new() },
tombstones: base.tombstones.clone(),
};
if let Err(e) = edit(Some(cur), &mut de) {
return EditOutcome { result: Err(EditError::Refused(e)), held: current };
}
let DocEdit { seq, entries, mut tombstones } = de;
let overlay = entries.overlay;
tombstones.retain(|_, t| seq.saturating_sub(t.seq) <= TOMBSTONE_KEEP_SEQS);
let built = super::chunk::rechunk_with(
&refs,
|i| Ok(Arc::new(super::docview::chunk_entries(&base.entries, &refs, i).map(|(k, e)| (k.clone(), e.clone())).collect())),
&overlay,
cfg.chunk_target,
cfg.chunk_min,
cfg.chunk_max,
)
.and_then(|chunks| {
let retired = super::docview::retired_over(cfg, &refs, &base.entries, &overlay);
if cfg.g_shadow || cfg!(debug_assertions) {
shadow_edit(cfg, base, &overlay, &chunks, &retired);
}
let tomb = match tombstones.is_empty() {
true => None,
false => Some(super::chunk::encode_tombstones(&tombstones)?.0),
};
Ok((chunks, retired, tomb))
});
let (chunks, retired, tomb) = match built {
Ok(b) => b,
Err(e) => return EditOutcome { result: Err(EditError::Store(e)), held: None },
};
let doc = ChunkedDoc { seq, sole_writer: false, boundary_source: None, tombstones: &tombstones };
let expected = cur.handle();
let sending = Pointer {
seq,
entries_key: None,
entries_seq: None,
chunks: Some(chunks.iter().map(|(r, _)| r.clone()).collect()),
sole_writer: false,
boundary_source: None,
epoch: 0,
tombstones: tomb,
};
match publish_chunk_list(store, cfg, &chunks, &doc, Some(&expected), &expected.prev_chunks, 0, flush).await {
Ok((meta, chunk_refs, pointer)) => {
debug_assert_eq!(pointer, sending, "the pointer published is not the one the own-write check compares");
log_retired(store, cfg, seq, 0, flush, retired, Some(&chunk_refs), Some(&expected)).await;
let mut held = current.expect("chunked above");
let l = Arc::make_mut(&mut held);
apply_overlay(&mut l.manifest.entries, overlay);
l.manifest.seq = seq;
l.manifest.sole_writer = false;
l.manifest.boundary_source = None;
l.manifest.tombstones = tombstones;
l.etag = meta.etag;
l.last_modified_unix = meta.last_modified_unix.or_else(|| Some(super::now_unix()));
let live: std::collections::HashSet<&str> =
pointer.chunks.iter().flatten().map(|r| r.addr.as_str()).collect();
cfg.alias_cache.lock().unwrap().retain(|a, _| live.contains(a.as_str()));
l.pointer = Some(pointer);
return EditOutcome { result: Ok(()), held: Some(held) };
}
Err(LeanError::Store(StoreError::PreconditionFailed(_))) | Err(LeanError::Store(StoreError::Conflict(_))) => {
sent = Some(sending);
continue;
}
Err(e) => return EditOutcome { result: Err(EditError::Store(e)), held: None },
}
}
EditOutcome { result: Err(EditError::Contended), held: None }
}
fn apply_overlay(entries: &mut BTreeMap<String, LeanEntry>, overlay: BTreeMap<String, Option<LeanEntry>>) {
for (p, v) in overlay {
match v {
Some(e) => {
entries.insert(p, e);
}
None => {
entries.remove(&p);
}
}
}
}
fn shadow_edit(
cfg: &LeanConfig,
base: &LeanManifest,
overlay: &BTreeMap<String, Option<LeanEntry>>,
chunks: &[(super::chunk::ChunkRef, Option<Vec<u8>>)],
retired: &[String],
) {
let mut whole = base.entries.clone();
apply_overlay(&mut whole, overlay.clone());
let want = super::chunk::split_with(&whole, cfg.chunk_target, cfg.chunk_min, cfg.chunk_max)
.expect("shadow: the whole document does not split");
let got_refs: Vec<&super::chunk::ChunkRef> = chunks.iter().map(|(r, _)| r).collect();
let want_refs: Vec<&super::chunk::ChunkRef> = want.iter().map(|(r, _)| r).collect();
assert_eq!(got_refs, want_refs, "edit shadow: the chunk list is not the whole document's");
for ((r, body), (_, w)) in chunks.iter().zip(&want) {
if let Some(b) = body {
assert_eq!(b, w, "edit shadow: chunk {} encodes differently", r.addr);
}
}
let still: std::collections::HashSet<&str> = whole.values().map(|e| e.key.as_str()).collect();
let want_retired: Vec<String> =
base.entries.values().map(|e| e.key.clone()).filter(|k| !still.contains(k.as_str())).collect();
assert_eq!(retired, want_retired.as_slice(), "edit shadow: the retire set is not the whole document's");
}
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 boundary_source = pub_stamps.boundary_source.map(|s| s.to_string()).or_else(|| m.boundary_source.clone());
let split = super::chunk::split_with(&m.entries, cfg.chunk_target, cfg.chunk_min, cfg.chunk_max)?;
let chunks: Vec<(super::chunk::ChunkRef, Option<Vec<u8>>)> = split.into_iter().map(|(r, b)| (r, Some(b))).collect();
let refs: Vec<super::chunk::ChunkRef> = chunks.iter().map(|(r, _)| r.clone()).collect();
super::docview::remember(cfg, &refs, &m.entries);
let doc = ChunkedDoc { seq: m.seq, sole_writer: m.sole_writer, boundary_source, tombstones: &m.tombstones };
let (meta, refs, _) =
publish_chunk_list(store, cfg, &chunks, &doc, expected, prev, pub_stamps.epoch, pub_stamps.flush_uuid).await?;
Ok((meta, refs))
}
pub struct ChunkedDoc<'a> {
pub seq: u64,
pub sole_writer: bool,
pub boundary_source: Option<String>,
pub tombstones: &'a BTreeMap<String, Tombstone>,
}
#[allow(clippy::too_many_arguments)]
async fn publish_chunk_list(
store: &dyn ObjectStore,
cfg: &LeanConfig,
chunks: &[(super::chunk::ChunkRef, Option<Vec<u8>>)],
doc: &ChunkedDoc<'_>,
expected: Option<&ManifestHandle>,
prev: &[super::chunk::ChunkRef],
epoch: u64,
flush_uuid: &str,
) -> LeanResult<(ObjectMeta, Vec<String>, Pointer)> {
let have: std::collections::HashSet<&str> = prev.iter().map(|r| r.addr.as_str()).collect();
let stamps = GenerationStamps {
generation: doc.seq,
epoch,
flush_uuid: flush_uuid.to_string(),
boundary_source: doc.boundary_source.clone(),
posix: None,
};
for (r, body) in chunks {
if have.contains(r.addr.as_str()) {
continue;
}
let Some(body) = body else {
return Err(LeanError::State(format!(
"chunk {} is not in the list this publish replaces and was never encoded",
r.addr
)));
};
let key = cfg.chunk_key(&r.addr);
let crc = crc64_nvme(body);
store.put_whole(&key, body.clone().into(), &PutCondition::Unconditional, &stamps, crc).await?;
if let Some(c) = cfg.chunk_cache.as_deref() {
c.put(&r.addr, body);
}
}
let tombstones = if doc.tombstones.is_empty() {
None
} else {
let (addr, body) = super::chunk::encode_tombstones(doc.tombstones)?;
let crc = crc64_nvme(&body);
if let Some(c) = cfg.chunk_cache.as_deref() {
c.put(&addr, &body);
}
store.put_whole(&cfg.chunk_key(&addr), body.into(), &PutCondition::Unconditional, &stamps, crc).await?;
Some(addr)
};
let pointer = Pointer {
seq: doc.seq,
entries_key: None,
entries_seq: None,
chunks: Some(chunks.iter().map(|(r, _)| r.clone()).collect()),
sole_writer: doc.sole_writer,
boundary_source: doc.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, pointer))
}
#[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 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();
log_retired(store, cfg, m.seq, epoch, flush_uuid, handles, refs.as_deref(), expected).await;
Ok(meta)
}
#[allow(clippy::too_many_arguments)]
pub async fn cas_write_view(
store: &dyn ObjectStore,
cfg: &LeanConfig,
chunks: &[(super::chunk::ChunkRef, Option<Vec<u8>>)],
doc: &ChunkedDoc<'_>,
expected: &ManifestHandle,
epoch: u64,
flush_uuid: &str,
retired_handles: Vec<String>,
) -> LeanResult<ObjectMeta> {
let (meta, refs, _) =
publish_chunk_list(store, cfg, chunks, doc, Some(expected), &expected.prev_chunks, epoch, flush_uuid).await?;
log_retired(store, cfg, doc.seq, epoch, flush_uuid, retired_handles, Some(&refs), Some(expected)).await;
Ok(meta)
}
#[allow(clippy::too_many_arguments)]
async fn log_retired(
store: &dyn ObjectStore,
cfg: &LeanConfig,
seq: u64,
epoch: u64,
flush_uuid: &str,
mut handles: Vec<String>,
refs: Option<&[String]>,
expected: Option<&ManifestHandle>,
) {
handles.sort();
handles.dedup();
let mut chunks: Vec<String> = vec![];
if let (Some(refs), Some(h)) = (refs, 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 = match serde_json::to_vec(&log) {
Ok(b) => b,
Err(e) => {
eprintln!("flint-sync: could not encode what seq {seq} retired ({e}); the sweeps judge it by write age");
return;
}
};
let crc = crc64_nvme(&body);
let stamps = GenerationStamps {
generation: 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 {seq} retired ({e}); the sweeps judge it by write age");
}
}
}
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 base_of = |p: &str| base.get(p).map(|b| b.as_str());
let at: BTreeMap<String, LeanEntry> = mine_upserts
.keys()
.chain(mine_deletes.iter())
.filter_map(|p| theirs.entries.get(p).map(|e| (p.clone(), e.clone())))
.collect();
let c = merge_changes(&base_of, &at, &theirs.tombstones, theirs.seq, mine_upserts, mine_deletes, parked);
let mut foreign: Vec<(String, LeanEntry)> = vec![];
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()));
}
}
let mut merged = theirs.clone();
merged.seq = c.seq;
merged.sole_writer = false;
merged.boundary_source = None;
for (p, v) in &c.overlay {
match v {
Some(e) => {
merged.entries.insert(p.clone(), e.clone());
}
None => {
merged.entries.remove(p);
}
}
}
merged.tombstones = c.tombstones;
Merged {
doc: merged,
foreign,
overridden: c.overridden,
deletes: c.deletes,
recreated: c.recreated,
over_theirs: c.over_theirs,
}
}
pub struct MergedChanges {
pub seq: u64,
pub overlay: BTreeMap<String, Option<LeanEntry>>,
pub tombstones: BTreeMap<String, Tombstone>,
pub overridden: Vec<(String, LeanEntry)>,
pub deletes: BTreeMap<String, DeleteOutcome>,
pub recreated: Vec<(String, String)>,
pub over_theirs: Vec<(String, LeanEntry)>,
}
pub fn merge_changes<'b>(
base: &dyn Fn(&str) -> Option<&'b str>,
theirs_at: &BTreeMap<String, LeanEntry>,
theirs_tombstones: &BTreeMap<String, Tombstone>,
theirs_seq: u64,
mine_upserts: &BTreeMap<String, LeanEntry>,
mine_deletes: &BTreeSet<String>,
parked: &BTreeSet<String>,
) -> MergedChanges {
let seq = theirs_seq + 1;
let mut tombstones = theirs_tombstones.clone();
let mut overlay: BTreeMap<String, Option<LeanEntry>> = BTreeMap::new();
let recreated: Vec<(String, String)> = mine_upserts
.keys()
.filter(|p| !theirs_at.contains_key(*p))
.filter_map(|p| {
let base_etag = base(p)?;
let deleted = theirs_tombstones.get(p).map(|t| t.etag.clone()).unwrap_or_else(|| base_etag.to_string());
Some((p.clone(), deleted))
})
.collect();
let overridden: Vec<(String, LeanEntry)> = mine_upserts
.keys()
.filter_map(|p| theirs_at.get(p).map(|e| (p, e)))
.filter(|(p, e)| base(p).map(|b| b != e.etag).unwrap_or(true))
.map(|(p, e)| (p.clone(), e.clone()))
.collect();
for (p, e) in mine_upserts {
overlay.insert(p.clone(), Some(e.clone()));
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_with(base(p), theirs_at.get(p), parked.contains(p));
deletes.insert(p.clone(), outcome);
if outcome == DeleteOutcome::OverTheirs {
if let Some(e) = theirs_at.get(p) {
over_theirs.push((p.clone(), e.clone()));
}
}
if outcome != DeleteOutcome::Parked {
let cited = match overlay.get(p) {
Some(v) => v.clone(),
None => theirs_at.get(p).cloned(),
};
if let Some(retired) = cited {
overlay.insert(p.clone(), None);
tombstones.insert(p.clone(), Tombstone { etag: retired.etag, seq });
}
}
}
tombstones.retain(|_, t| seq.saturating_sub(t.seq) <= TOMBSTONE_KEEP_SEQS);
MergedChanges { seq, overlay, tombstones, 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 {
delete_outcome_with(base.get(p).map(|b| b.as_str()), theirs.entries.get(p), parked.contains(p))
}
pub fn delete_outcome_with(base: Option<&str>, theirs: Option<&LeanEntry>, parked: bool) -> DeleteOutcome {
if parked {
return DeleteOutcome::Parked;
}
match (theirs, base) {
(Some(e), Some(b)) if e.etag == b => DeleteOutcome::Applied,
(None, _) => DeleteOutcome::Applied,
(Some(_), _) => DeleteOutcome::OverTheirs,
}
}