pub mod barrier;
pub mod checkout;
pub mod chunk;
pub mod control;
pub mod gauges;
pub use gauges::{status_report, Gauges, StatusReport};
pub mod inbox;
pub mod lease;
pub mod manifest;
pub mod metrics;
pub mod reader;
pub mod scan;
pub(crate) mod safefs;
pub mod sentinel;
pub mod state;
pub mod sync;
pub mod trace;
pub mod uds;
pub mod verbs;
#[cfg(test)]
mod tests;
#[cfg(test)] mod tests_upload_gate;
use std::path::PathBuf;
use std::sync::Arc;
use flint_store::ObjectStore;
pub const LEAN_DIR: &str = ".flint/lean";
pub const STATE_DIR: &str = ".flint-sync";
pub const CONTROL_DIR: &str = ".flint";
pub const SENTINEL_PROTOCOL: u32 = 1;
pub const SYNCER_VERSION: &str = env!("CARGO_PKG_VERSION");
pub const WHOLE_PUT_MAX: u64 = 64 * 1024 * 1024;
#[derive(Debug, thiserror::Error)]
pub enum LeanError {
#[error("store: {0}")]
Store(#[from] flint_store::StoreError),
#[error("io: {0}")]
Io(#[from] std::io::Error),
#[error("state: {0}")]
State(String),
#[error("fenced: {0}")]
Fenced(String),
#[error("budget: {0}")]
Budget(String),
#[error("refused: {0}")]
Refused(String),
}
pub const EXIT_REFUSED: i32 = 78;
impl LeanError {
pub fn is_auth(&self) -> bool {
matches!(self, LeanError::Store(flint_store::StoreError::Auth(_)))
}
}
pub type LeanResult<T> = Result<T, LeanError>;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SentinelMode {
Auto,
Off,
Force,
}
impl SentinelMode {
pub fn parse(s: &str) -> Option<SentinelMode> {
match s {
"auto" => Some(SentinelMode::Auto),
"off" => Some(SentinelMode::Off),
"force" => Some(SentinelMode::Force),
_ => None,
}
}
pub fn as_str(&self) -> &'static str {
match self {
SentinelMode::Auto => "auto",
SentinelMode::Off => "off",
SentinelMode::Force => "force",
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
pub enum Access {
#[default]
ReadWrite,
Read,
}
impl Access {
pub fn parse(s: &str) -> Option<Access> {
match s {
"read" => Some(Access::Read),
"readWrite" => Some(Access::ReadWrite),
_ => None,
}
}
pub fn as_str(&self) -> &'static str {
match self {
Access::ReadWrite => "readWrite",
Access::Read => "read",
}
}
pub fn is_read(&self) -> bool {
*self == Access::Read
}
}
#[derive(Clone)]
pub struct LeanConfig {
pub prefix: String,
pub root: PathBuf,
pub floor_secs: u64,
pub sole_writer: bool,
pub access: Access,
pub whole_put_max: u64,
pub chunked: bool,
pub orphan_grace_secs: u64,
pub chunk_target: usize,
pub chunk_min: usize,
pub chunk_max: usize,
pub max_bytes: u64,
pub max_files: u64,
pub window_slack_secs: u64,
pub fanout: usize,
pub upload_fanout: usize,
pub project_id: Option<String>,
pub fetch_inflight_max_bytes: u64,
pub fetch_drivers: usize,
pub range_get_min_bytes: u64,
pub range_get_chunk_bytes: u64,
pub range_get_parallelism: usize,
pub sentinel_mode: SentinelMode,
pub sentinel_min_interval_secs: u64,
pub sentinel_hourly_budget: u64,
pub sentinel_poll_secs: u64,
pub claim_poll_secs: u64,
pub claim_poll_head_ms: u64,
pub claim_quiet_spacing_secs: u64,
pub claim_deadline_secs: u64,
pub drill_hold_commit_secs: u64,
pub drill_hold_gc_secs: u64,
pub event_trace: Option<trace::Sink>,
}
impl LeanConfig {
pub fn new(prefix: &str, root: impl Into<PathBuf>) -> Self {
LeanConfig {
prefix: prefix.trim_end_matches('/').to_string(),
root: root.into(),
floor_secs: 60,
sole_writer: false,
access: Access::ReadWrite,
whole_put_max: WHOLE_PUT_MAX,
chunked: true,
orphan_grace_secs: manifest::ORPHAN_GRACE_SECS,
chunk_target: chunk::CHUNK_TARGET,
chunk_min: chunk::CHUNK_MIN,
chunk_max: chunk::CHUNK_MAX,
max_bytes: 0,
max_files: 0,
window_slack_secs: 180,
fanout: 128,
upload_fanout: 32,
project_id: None,
fetch_inflight_max_bytes: 128 * 1024 * 1024,
fetch_drivers: std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1)
.clamp(1, 8),
range_get_min_bytes: 8 * 1024 * 1024,
range_get_chunk_bytes: 16 * 1024 * 1024,
range_get_parallelism: 4,
sentinel_mode: SentinelMode::Auto,
sentinel_min_interval_secs: 5,
sentinel_hourly_budget: 60,
sentinel_poll_secs: 1,
claim_poll_secs: lease::CLAIM_POLL_SECS,
claim_poll_head_ms: lease::CLAIM_POLL_HEAD_MS,
claim_quiet_spacing_secs: lease::QUIET_SPACING_SECS,
claim_deadline_secs: lease::CLAIM_DEADLINE_SECS,
drill_hold_commit_secs: 0,
drill_hold_gc_secs: 0,
event_trace: None,
}
}
pub fn file_key(&self, path: &str) -> String {
format!("{}/files/{}", self.prefix, path)
}
pub fn manifest_key(&self) -> String {
format!("{}/{}/manifest", self.prefix, LEAN_DIR)
}
pub fn current_key(&self) -> String {
format!("{}/{}/current", self.prefix, LEAN_DIR)
}
pub fn generation_key(&self, seq: u64, flush_uuid: &str) -> String {
format!("{}/{}/manifests/{seq:020}-{flush_uuid}", self.prefix, LEAN_DIR)
}
pub fn chunk_key(&self, addr: &str) -> String {
format!("{}/{}/chunks/{addr}", self.prefix, LEAN_DIR)
}
pub fn inbox_key(&self) -> String {
format!("{}/{}/inbox", self.prefix, LEAN_DIR)
}
pub fn epoch_key(&self) -> String {
format!("{}/{}/epoch", self.prefix, LEAN_DIR)
}
pub fn claim_key(&self) -> String {
format!("{}/{}/claim", self.prefix, LEAN_DIR)
}
pub fn conflict_key(&self, uuid: &str, path: &str) -> String {
format!("{}/{}/conflicts/{}/{}", self.prefix, LEAN_DIR, uuid, path)
}
pub fn draft_body_key(&self, user: &str, path: &str) -> String {
format!("{}/{}/drafts/{}/body/{}", self.prefix, LEAN_DIR, user, path)
}
pub fn draft_meta_key(&self, user: &str, path: &str) -> String {
format!("{}/{}/drafts/{}/meta/{}", self.prefix, LEAN_DIR, user, path)
}
pub fn draft_meta_prefix(&self, user: &str) -> String {
format!("{}/{}/drafts/{}/meta/", self.prefix, LEAN_DIR, user)
}
pub fn state_dir(&self) -> PathBuf {
self.root.join(STATE_DIR)
}
pub fn control_dir(&self) -> PathBuf {
self.root.join(CONTROL_DIR)
}
}
pub struct Syncer {
pub store: Arc<dyn ObjectStore>,
pub cfg: LeanConfig,
pub state: state::SyncerState,
pub lease: Option<flint_store::EpochLease>,
pub noted_not_regular: std::collections::BTreeSet<String>,
}
pub fn now_unix() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}