use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::sync::{Arc, Mutex};
pub struct FeltDb;
pub const STATE_MODEL_VERSION: u32 = 1;
#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
pub struct StateId(String);
impl StateId {
pub fn compute(canonical_json: &str) -> Self {
let mut hasher = Sha256::new();
hasher.update(canonical_json.as_bytes());
let result = hasher.finalize();
StateId(format!("{:x}", result))
}
pub fn as_hex(&self) -> &str {
&self.0
}
pub fn from_hex(hex: String) -> Self {
StateId(hex)
}
}
impl std::fmt::Display for StateId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.0)
}
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct StateRevision {
pub id: StateId,
pub resource: String,
pub content: String,
pub content_id: StateId,
pub parent_id: Option<StateId>,
pub sequence: u64,
pub authority: String,
pub timestamp_ms: u64,
pub metadata: BTreeMap<String, Value>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct StateObject {
pub id: StateId,
pub content: String,
#[serde(default)]
pub metadata: BTreeMap<String, Value>,
}
impl StateObject {
pub fn new(content: String) -> Self {
let id = StateId::compute(&content);
Self {
id,
content,
metadata: BTreeMap::new(),
}
}
pub fn verify_integrity(&self) -> bool {
self.id == StateId::compute(&self.content)
}
}
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct StateRef {
pub namespace: String,
pub name: String,
pub target: StateId,
pub version: u64,
pub authority: String,
pub updated_at_ms: u64,
#[serde(default)]
pub metadata: BTreeMap<String, Value>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RefError {
InvalidIdentifier,
NotFound,
AlreadyExists,
UnknownTarget(StateId),
VersionConflict { expected: u64, current: u64 },
Storage(String),
}
impl std::fmt::Display for RefError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
RefError::InvalidIdentifier => write!(f, "invalid ref identifier"),
RefError::NotFound => write!(f, "ref not found"),
RefError::AlreadyExists => write!(f, "ref already exists"),
RefError::UnknownTarget(id) => write!(f, "ref target {id} does not exist"),
RefError::VersionConflict { expected, current } => {
write!(f, "stale ref version: expected {expected}, current {current}")
}
RefError::Storage(message) => write!(f, "{message}"),
}
}
}
impl std::error::Error for RefError {}
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct MaterializationProjection {
#[serde(default)]
pub paths: Vec<Vec<PathComponent>>,
}
impl MaterializationProjection {
pub fn full() -> Self {
Self { paths: Vec::new() }
}
pub fn select(paths: Vec<Vec<PathComponent>>) -> Self {
Self { paths }
}
pub fn is_full(&self) -> bool {
self.paths.is_empty() || self.paths.iter().any(|path| path.is_empty())
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum MaterializationError {
RevisionNotFound(StateId),
ContentUnparseable(String),
MissingPath(Vec<PathComponent>),
NonDeterministicProjection,
}
impl std::fmt::Display for MaterializationError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
MaterializationError::RevisionNotFound(id) => write!(f, "revision {id} not found"),
MaterializationError::ContentUnparseable(error) => {
write!(f, "revision content is not parseable JSON: {error}")
}
MaterializationError::MissingPath(path) => {
write!(f, "projection path does not resolve: {path:?}")
}
MaterializationError::NonDeterministicProjection => write!(
f,
"projection contains overlapping paths and does not deterministically resolve"
),
}
}
}
impl std::error::Error for MaterializationError {}
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct GarbageCollectionReport {
pub roots: Vec<StateRef>,
pub reachable_revisions: Vec<StateId>,
pub reachable_objects: Vec<StateId>,
pub collected_revisions: Vec<StateId>,
pub collected_objects: Vec<StateId>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum GarbageCollectionError {
Storage(String),
}
impl std::fmt::Display for GarbageCollectionError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
GarbageCollectionError::Storage(message) => write!(f, "{message}"),
}
}
}
impl std::error::Error for GarbageCollectionError {}
fn compute_revision_id(
resource: &str,
content_id: &StateId,
parent_id: Option<&StateId>,
sequence: u64,
) -> StateId {
let parent = parent_id.map(StateId::as_hex).unwrap_or("");
let material = format!(
"feltdb.revision.v2\n{}:{}\n{}:{}\n{}:{}\n{}",
resource.len(),
resource,
content_id.as_hex().len(),
content_id.as_hex(),
parent.len(),
parent,
sequence,
);
StateId::compute(&material)
}
impl StateRevision {
pub fn initial(resource: String, content: String, authority: String) -> Self {
Self::at(resource, content, None, 0, authority)
}
pub fn child(content: String, parent: &StateRevision, authority: String) -> Self {
Self::at(
parent.resource.clone(),
content,
Some(parent.id.clone()),
parent.sequence + 1,
authority,
)
}
pub fn at(
resource: String,
content: String,
parent_id: Option<StateId>,
sequence: u64,
authority: String,
) -> Self {
let content_id = StateId::compute(&content);
let id = compute_revision_id(&resource, &content_id, parent_id.as_ref(), sequence);
let timestamp_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64;
StateRevision {
id,
resource,
content,
content_id,
parent_id,
sequence,
authority,
timestamp_ms,
metadata: BTreeMap::new(),
}
}
pub fn verify_integrity(&self) -> bool {
self.content_id == StateId::compute(&self.content)
&& self.id
== compute_revision_id(
&self.resource,
&self.content_id,
self.parent_id.as_ref(),
self.sequence,
)
}
pub fn parent(&self) -> Option<&StateId> {
self.parent_id.as_ref()
}
pub fn equals(&self, other: &StateRevision) -> bool {
self.id == other.id
}
pub fn same_content_as(&self, other: &StateRevision) -> bool {
self.content_id == other.content_id
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum Relationship {
Ancestor,
Descendant,
Identity,
Diverged,
Unrelated,
}
pub struct StateTopology {
revisions: HashMap<StateId, StateRevision>,
parents: HashMap<StateId, StateId>,
}
impl StateTopology {
pub fn new() -> Self {
StateTopology {
revisions: HashMap::new(),
parents: HashMap::new(),
}
}
pub fn add_revision(&mut self, revision: StateRevision) {
if let Some(parent_id) = revision.parent_id.clone() {
self.parents.insert(revision.id.clone(), parent_id);
}
self.revisions.insert(revision.id.clone(), revision);
}
pub fn get(&self, id: &StateId) -> Option<&StateRevision> {
self.revisions.get(id)
}
pub fn is_ancestor(&self, a: &StateId, b: &StateId) -> bool {
if a == b {
return true;
}
let mut visited = BTreeSet::new();
let mut current = b.clone();
while let Some(parent_id) = self.parents.get(¤t) {
if parent_id == a {
return true;
}
if !visited.insert(current.clone()) {
return false;
}
current = parent_id.clone();
}
false
}
pub fn ancestors(&self, id: &StateId) -> Vec<StateId> {
let mut ancestors = Vec::new();
let mut visited = BTreeSet::new();
let mut current = id.clone();
while let Some(parent_id) = self.parents.get(¤t) {
if !visited.insert(current.clone()) {
break;
}
ancestors.push(parent_id.clone());
current = parent_id.clone();
}
ancestors
}
pub fn common_ancestor(&self, a: &StateId, b: &StateId) -> Option<StateId> {
if a == b {
return Some(a.clone());
}
let ancestors_a: BTreeSet<_> = self.ancestors(a).iter().cloned().collect();
let mut visited = BTreeSet::new();
let mut current = b.clone();
while let Some(parent_id) = self.parents.get(¤t) {
if ancestors_a.contains(parent_id) {
return Some(parent_id.clone());
}
if !visited.insert(current.clone()) {
break;
}
current = parent_id.clone();
}
None
}
pub fn relationship(&self, a: &StateId, b: &StateId) -> Relationship {
if a == b {
return Relationship::Identity;
}
if self.is_ancestor(a, b) {
return Relationship::Ancestor;
}
if self.is_ancestor(b, a) {
return Relationship::Descendant;
}
if self.common_ancestor(a, b).is_some() {
return Relationship::Diverged;
}
Relationship::Unrelated
}
}
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub enum ChangeKind {
Added,
Removed,
Changed,
}
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub enum PathComponent {
Key(String),
Index(usize),
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum PathRelation {
Same,
Ancestor,
Descendant,
Disjoint,
}
fn path_starts_with(candidate: &[PathComponent], prefix: &[PathComponent]) -> bool {
candidate.len() >= prefix.len() && candidate[..prefix.len()] == *prefix
}
pub fn path_relation(one: &[PathComponent], two: &[PathComponent]) -> PathRelation {
if one == two {
PathRelation::Same
} else if path_starts_with(two, one) {
PathRelation::Ancestor
} else if path_starts_with(one, two) {
PathRelation::Descendant
} else {
PathRelation::Disjoint
}
}
pub fn paths_overlap(one: &[PathComponent], two: &[PathComponent]) -> bool {
!matches!(path_relation(one, two), PathRelation::Disjoint)
}
pub fn resolve_path<'a>(value: &'a Value, path: &[PathComponent]) -> Option<&'a Value> {
let mut current = value;
for component in path {
current = match component {
PathComponent::Key(key) => current.as_object()?.get(key)?,
PathComponent::Index(index) => current.as_array()?.get(*index)?,
};
}
Some(current)
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SemanticChange {
pub path: Vec<PathComponent>,
pub kind: ChangeKind,
pub old_value: Option<Value>,
pub new_value: Option<Value>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct SemanticDiff {
pub changes: Vec<SemanticChange>,
pub is_read_only: bool,
}
impl SemanticDiff {
pub fn compute(old: &Value, new: &Value) -> Self {
let mut changes = Vec::new();
Self::diff_recursive(old, new, &mut vec![], &mut changes);
changes.sort_by(|a, b| a.path.cmp(&b.path).then_with(|| a.kind.cmp(&b.kind)));
SemanticDiff {
changes,
is_read_only: false,
}
}
fn diff_recursive(
old: &Value,
new: &Value,
path: &mut Vec<PathComponent>,
changes: &mut Vec<SemanticChange>,
) {
use serde_json::{json, Value::*};
match (old, new) {
(Null, Null) => {}
(Bool(a), Bool(b)) if a == b => {}
(Number(a), Number(b)) if a == b => {}
(String(a), String(b)) if a == b => {}
(Array(a), Array(b)) => {
for (i, (old_v, new_v)) in a.iter().zip(b.iter()).enumerate() {
path.push(PathComponent::Index(i));
Self::diff_recursive(old_v, new_v, path, changes);
path.pop();
}
for (i, new_v) in b.iter().enumerate().skip(a.len()) {
path.push(PathComponent::Index(i));
changes.push(SemanticChange {
path: path.clone(),
kind: ChangeKind::Added,
old_value: None,
new_value: Some(new_v.clone()),
});
path.pop();
}
for (i, old_v) in a.iter().enumerate().skip(b.len()) {
path.push(PathComponent::Index(i));
changes.push(SemanticChange {
path: path.clone(),
kind: ChangeKind::Removed,
old_value: Some(old_v.clone()),
new_value: None,
});
path.pop();
}
}
(Object(a), Object(b)) => {
for key in a.keys() {
if let Some(new_v) = b.get(key) {
path.push(PathComponent::Key(key.clone()));
Self::diff_recursive(&a[key], new_v, path, changes);
path.pop();
} else {
path.push(PathComponent::Key(key.clone()));
changes.push(SemanticChange {
path: path.clone(),
kind: ChangeKind::Removed,
old_value: Some(a[key].clone()),
new_value: None,
});
path.pop();
}
}
for key in b.keys() {
if !a.contains_key(key) {
path.push(PathComponent::Key(key.clone()));
changes.push(SemanticChange {
path: path.clone(),
kind: ChangeKind::Added,
old_value: None,
new_value: Some(b[key].clone()),
});
path.pop();
}
}
}
(_, _) => {
changes.push(SemanticChange {
path: path.clone(),
kind: ChangeKind::Changed,
old_value: Some(old.clone()),
new_value: Some(new.clone()),
});
}
}
}
pub fn read_only(mut self) -> Self {
self.is_read_only = true;
self
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum ConflictClass {
Independent,
Convergent,
Conflict,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct PathConflict {
pub path: Vec<PathComponent>,
pub classification: ConflictClass,
pub base_value: Option<Value>,
pub left_value: Option<Value>,
pub right_value: Option<Value>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ConflictClassification {
pub path_conflicts: Vec<PathConflict>,
pub overall: ConflictClass,
}
impl ConflictClassification {
pub fn classify(base: &StateRevision, left: &StateRevision, right: &StateRevision) -> Self {
use serde_json::Value;
let base_json: Value = serde_json::from_str(&base.content).unwrap_or(Value::Null);
let left_json: Value = serde_json::from_str(&left.content).unwrap_or(Value::Null);
let right_json: Value = serde_json::from_str(&right.content).unwrap_or(Value::Null);
let left_diff = SemanticDiff::compute(&base_json, &left_json);
let right_diff = SemanticDiff::compute(&base_json, &right_json);
let mut path_conflicts = Vec::new();
let mut has_conflict = false;
let left_paths: BTreeMap<Vec<PathComponent>, &SemanticChange> = left_diff
.changes
.iter()
.map(|c| (c.path.clone(), c))
.collect();
let right_paths: BTreeMap<Vec<PathComponent>, &SemanticChange> = right_diff
.changes
.iter()
.map(|c| (c.path.clone(), c))
.collect();
let all_paths: BTreeSet<_> = left_paths
.keys()
.chain(right_paths.keys())
.cloned()
.collect();
for path in all_paths {
let left_change = left_paths.get(&path);
let right_change = right_paths.get(&path);
let classification = match (left_change, right_change) {
(Some(lc), Some(rc)) => {
if lc.kind == rc.kind && lc.new_value == rc.new_value {
ConflictClass::Convergent
} else {
has_conflict = true;
ConflictClass::Conflict
}
}
(Some(_), None) => {
if overlapping_change_exists(&right_paths, &path) {
has_conflict = true;
ConflictClass::Conflict
} else {
ConflictClass::Independent
}
}
(None, Some(_)) => {
if overlapping_change_exists(&left_paths, &path) {
has_conflict = true;
ConflictClass::Conflict
} else {
ConflictClass::Independent
}
}
(None, None) => ConflictClass::Independent,
};
path_conflicts.push(PathConflict {
path: path.clone(),
classification,
base_value: resolve_path(&base_json, &path).cloned(),
left_value: left_change.and_then(|c| c.new_value.clone()),
right_value: right_change.and_then(|c| c.new_value.clone()),
});
}
path_conflicts.sort_by(|a, b| a.path.cmp(&b.path));
let overall = if has_conflict {
ConflictClass::Conflict
} else if path_conflicts
.iter()
.any(|c| c.classification == ConflictClass::Convergent)
{
ConflictClass::Convergent
} else {
ConflictClass::Independent
};
ConflictClassification {
path_conflicts,
overall,
}
}
}
fn overlapping_change_exists(
changes: &BTreeMap<Vec<PathComponent>, &SemanticChange>,
path: &[PathComponent],
) -> bool {
for depth in 0..path.len() {
if changes.contains_key(&path[..depth].to_vec()) {
return true;
}
}
changes
.range(path.to_vec()..)
.find(|(candidate, _)| candidate.as_slice() != path)
.is_some_and(|(candidate, _)| path_starts_with(candidate, path))
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ReconciliationPlan {
pub left_id: StateId,
pub right_id: StateId,
pub base_id: StateId,
pub parent_choice: bool,
pub path_overrides: BTreeMap<Vec<PathComponent>, Value>,
}
impl ReconciliationPlan {
pub fn new(left_id: StateId, right_id: StateId, base_id: StateId, parent_choice: bool) -> Self {
ReconciliationPlan {
left_id,
right_id,
base_id,
parent_choice,
path_overrides: BTreeMap::new(),
}
}
pub fn with_override(mut self, path: Vec<PathComponent>, value: Value) -> Self {
self.path_overrides.insert(path, value);
self
}
pub fn validate(&self) -> bool {
!self.left_id.as_hex().is_empty()
&& !self.right_id.as_hex().is_empty()
&& !self.base_id.as_hex().is_empty()
}
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct StateReconciliationResult {
pub materialized_state: StateRevision,
pub plan: ReconciliationPlan,
}
pub const RECONCILE_PLAN_INVALID: &str = "RECONCILE_PLAN_INVALID";
pub const RECONCILE_REVISION_MISMATCH: &str = "RECONCILE_REVISION_MISMATCH";
pub const RECONCILE_CONTENT_UNPARSEABLE: &str = "RECONCILE_CONTENT_UNPARSEABLE";
pub const RECONCILE_OVERRIDE_PATHS_OVERLAP: &str = "RECONCILE_OVERRIDE_PATHS_OVERLAP";
pub const RECONCILE_OVERRIDE_UNADDRESSABLE: &str = "RECONCILE_OVERRIDE_UNADDRESSABLE";
impl ReconciliationPlan {
pub fn selected_parent<'a>(
&self,
left: &'a StateRevision,
right: &'a StateRevision,
) -> &'a StateRevision {
if self.parent_choice {
left
} else {
right
}
}
pub fn unresolved_conflicts(
&self,
classification: &ConflictClassification,
) -> Vec<Vec<PathComponent>> {
classification
.path_conflicts
.iter()
.filter(|entry| entry.classification == ConflictClass::Conflict)
.map(|entry| entry.path.clone())
.filter(|path| !self.path_overrides.contains_key(path))
.collect()
}
}
pub fn apply_reconciliation_plan(
base: &StateRevision,
left: &StateRevision,
right: &StateRevision,
plan: &ReconciliationPlan,
) -> Result<String, String> {
if !plan.validate() {
return Err(format!(
"{RECONCILE_PLAN_INVALID}: the plan does not reference three revisions"
));
}
for (label, expected, supplied) in [
("base", &plan.base_id, &base.id),
("left", &plan.left_id, &left.id),
("right", &plan.right_id, &right.id),
] {
if expected != supplied {
return Err(format!(
"{RECONCILE_REVISION_MISMATCH}: the plan names a different {label} revision"
));
}
}
let parent = plan.selected_parent(left, right);
let mut state: Value = serde_json::from_str(&parent.content).map_err(|error| {
format!(
"{RECONCILE_CONTENT_UNPARSEABLE}: the selected parent's content is not JSON: {error}"
)
})?;
let paths: Vec<&Vec<PathComponent>> = plan.path_overrides.keys().collect();
for pair in paths.windows(2) {
if paths_overlap(pair[0], pair[1]) {
return Err(format!(
"{RECONCILE_OVERRIDE_PATHS_OVERLAP}: {:?} and {:?} overlap, so the plan does \
not say what the result is",
pair[0], pair[1]
));
}
}
for (path, value) in &plan.path_overrides {
assign_path(&mut state, path, value.clone())?;
}
serde_json::to_string(&state)
.map_err(|error| format!("{RECONCILE_CONTENT_UNPARSEABLE}: {error}"))
}
fn assign_path(state: &mut Value, path: &[PathComponent], value: Value) -> Result<(), String> {
let Some((last, prefix)) = path.split_last() else {
*state = value;
return Ok(());
};
let mut current = state;
for (depth, component) in prefix.iter().enumerate() {
current = match component {
PathComponent::Key(key) => current.as_object_mut().and_then(|o| o.get_mut(key)),
PathComponent::Index(index) => current.as_array_mut().and_then(|a| a.get_mut(*index)),
}
.ok_or_else(|| {
format!(
"{RECONCILE_OVERRIDE_UNADDRESSABLE}: {path:?} does not resolve at \
component {depth} ({component:?})"
)
})?;
}
match last {
PathComponent::Key(key) => match current.as_object_mut() {
Some(members) => {
members.insert(key.clone(), value);
Ok(())
}
None => Err(format!(
"{RECONCILE_OVERRIDE_UNADDRESSABLE}: {path:?} names a member of \
something that is not an object"
)),
},
PathComponent::Index(index) => match current.as_array_mut() {
Some(items) if *index < items.len() => {
items[*index] = value;
Ok(())
}
Some(items) => Err(format!(
"{RECONCILE_OVERRIDE_UNADDRESSABLE}: {path:?} is past the end of an \
array of length {}",
items.len()
)),
None => Err(format!(
"{RECONCILE_OVERRIDE_UNADDRESSABLE}: {path:?} indexes something that \
is not an array"
)),
},
}
}
fn assign_projection(
state: &mut Value,
path: &[PathComponent],
value: Value,
) -> Result<(), MaterializationError> {
let Some((last, prefix)) = path.split_last() else {
*state = value;
return Ok(());
};
let mut current = state;
for component in prefix {
match component {
PathComponent::Key(key) => {
if current.is_null() {
*current = Value::Object(serde_json::Map::new());
}
let Some(object) = current.as_object_mut() else {
return Err(MaterializationError::MissingPath(path.to_vec()));
};
current = object.entry(key.clone()).or_insert(Value::Null);
}
PathComponent::Index(index) => {
if current.is_null() {
*current = Value::Array(Vec::new());
}
let Some(array) = current.as_array_mut() else {
return Err(MaterializationError::MissingPath(path.to_vec()));
};
while array.len() <= *index {
array.push(Value::Null);
}
current = &mut array[*index];
}
}
}
match last {
PathComponent::Key(key) => {
if current.is_null() {
*current = Value::Object(serde_json::Map::new());
}
let Some(object) = current.as_object_mut() else {
return Err(MaterializationError::MissingPath(path.to_vec()));
};
object.insert(key.clone(), value);
Ok(())
}
PathComponent::Index(index) => {
if current.is_null() {
*current = Value::Array(Vec::new());
}
let Some(array) = current.as_array_mut() else {
return Err(MaterializationError::MissingPath(path.to_vec()));
};
while array.len() <= *index {
array.push(Value::Null);
}
array[*index] = value;
Ok(())
}
}
}
pub fn reconcile(
base: &StateRevision,
left: &StateRevision,
right: &StateRevision,
plan: &ReconciliationPlan,
authority: String,
) -> Result<StateReconciliationResult, String> {
let content = apply_reconciliation_plan(base, left, right, plan)?;
let parent = plan.selected_parent(left, right);
Ok(StateReconciliationResult {
materialized_state: StateRevision::child(content, parent, authority),
plan: plan.clone(),
})
}
pub struct StateStore {
backing: Backing,
mutation_lock: Arc<Mutex<()>>,
}
enum Backing {
Volatile(Arc<Mutex<VolatileState>>),
Durable(Arc<crate::FeltDb>),
}
#[derive(Default)]
struct VolatileState {
revisions: HashMap<StateId, StateRevision>,
retention: HashMap<String, RetentionState>,
refs: HashMap<(String, String), StateRef>,
objects: HashMap<StateId, StateObject>,
}
impl Clone for Backing {
fn clone(&self) -> Self {
match self {
Backing::Volatile(map) => Backing::Volatile(map.clone()),
Backing::Durable(db) => Backing::Durable(db.clone()),
}
}
}
fn revision_key(id: &StateId) -> String {
format!("state:revision:{}", id.as_hex())
}
fn object_key(id: &StateId) -> String {
format!("state:object:{}", id.as_hex())
}
fn ref_key(namespace: &str, name: &str) -> String {
format!("state:ref:{}:{}", encode_ref_part(namespace), encode_ref_part(name))
}
fn encode_ref_part(value: &str) -> String {
value.replace('%', "%25").replace(':', "%3A")
}
fn valid_ref_component(value: &str) -> bool {
!value.trim().is_empty()
}
fn looks_like_state_id(value: &str) -> bool {
value.len() == 64 && value.chars().all(|c| c.is_ascii_hexdigit())
}
fn current_time_ms() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64
}
pub(crate) const GC_GRACE_MS: u64 = 15 * 60 * 1000;
pub(crate) mod durable {
use super::*;
use crate::{AtomicMutation, FeltDb, FlowError, PreconditionFailure, RecordPrecondition};
use serde_json::json;
use std::sync::atomic::{AtomicU64, Ordering};
const ATTEMPTS: usize = 64;
pub(super) const GENERATION_KEY: &str = "state:_gcgen";
fn incarnation_key(namespace: &str, name: &str) -> String {
format!(
"state:refinc:{}:{}",
encode_ref_part(namespace),
encode_ref_part(name)
)
}
fn put(key: &str, value: Value) -> AtomicMutation {
AtomicMutation {
capability: "state".into(),
key: key.into(),
value: Some(value),
}
}
fn remove(key: &str) -> AtomicMutation {
AtomicMutation {
capability: "state".into(),
key: key.into(),
value: None,
}
}
fn row_version(row: &Value) -> u64 {
row.get("__version").and_then(Value::as_u64).unwrap_or(1)
}
fn unchanged(key: &str, read: Option<&Value>) -> RecordPrecondition {
RecordPrecondition {
capability: "state".into(),
key: key.into(),
require_absent: read.is_none(),
expected_version: read.map(row_version),
..RecordPrecondition::default()
}
}
fn present(key: &str) -> RecordPrecondition {
RecordPrecondition {
capability: "state".into(),
key: key.into(),
expected_version: Some(1),
..RecordPrecondition::default()
}
}
fn advance_generation(transaction_id: &str) -> AtomicMutation {
let digest = Sha256::digest(transaction_id.as_bytes());
let mut bytes = [0u8; 8];
bytes.copy_from_slice(&digest[..8]);
put(
GENERATION_KEY,
json!({ "__version": u64::from_be_bytes(bytes), "transaction": transaction_id }),
)
}
fn unique_suffix() -> String {
static COUNTER: AtomicU64 = AtomicU64::new(0);
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
format!(
"{}-{nanos}-{}",
std::process::id(),
COUNTER.fetch_add(1, Ordering::Relaxed)
)
}
fn ref_transaction_id(
kind: &str,
namespace: &str,
name: &str,
incarnation: u64,
version: Option<u64>,
) -> String {
let mut hasher = Sha256::new();
for part in [
"feltdb.state.ref.v1",
kind,
namespace,
name,
&incarnation.to_string(),
&version.map(|v| v.to_string()).unwrap_or_default(),
] {
hasher.update(part.as_bytes());
hasher.update([0u8]);
}
format!("state-ref:{:x}", hasher.finalize())
}
fn payload_hash(payload: &Value) -> String {
format!("sha256:{:x}", Sha256::digest(payload.to_string().as_bytes()))
}
fn incarnation_of(row: &Value) -> u64 {
row.get("__incarnation").and_then(Value::as_u64).unwrap_or(0)
}
fn incarnation_in(row: Option<&Value>) -> u64 {
row.and_then(|row| row.get("incarnation").and_then(Value::as_u64))
.unwrap_or(0)
}
fn ref_row(reference: &StateRef, incarnation: u64) -> Value {
let mut row = serde_json::to_value(reference).unwrap_or(Value::Null);
if let Some(object) = row.as_object_mut() {
object.insert("__incarnation".into(), Value::from(incarnation));
}
row
}
enum Refusal {
Changed,
OtherRequest,
Failed(String),
}
fn refusal(error: FlowError) -> Refusal {
match error {
FlowError::PreconditionFailed(failure) => match *failure {
PreconditionFailure::Present { .. }
| PreconditionFailure::Missing { .. }
| PreconditionFailure::Version { .. } => Refusal::Changed,
other => Refusal::Failed(format!("{other:?}")),
},
FlowError::TransactionIdPayloadMismatch { .. } => Refusal::OtherRequest,
other => Refusal::Failed(other.to_string()),
}
}
fn storage(error: impl std::fmt::Display) -> RefError {
RefError::Storage(error.to_string())
}
fn contended(what: &str) -> String {
format!(
"{what} was refused by concurrent changes on {ATTEMPTS} consecutive attempts; nothing was applied"
)
}
fn row(db: &FeltDb, key: &str) -> Result<Option<Value>, RefError> {
db.get::<Value>(key).map_err(storage)
}
fn committed_payload(db: &FeltDb, id: &str) -> Result<Option<String>, RefError> {
db.applied_transaction_payload_hash(id).map_err(storage)
}
fn decode(row: Value) -> Result<StateRef, RefError> {
serde_json::from_value(row).map_err(storage)
}
fn commit(
db: &FeltDb,
id: &str,
payload: Option<&str>,
pins: &[RecordPrecondition],
mutations: &[AtomicMutation],
) -> Result<(), Refusal> {
db.apply_atomic_transaction_guarded(id, payload, None, &[], pins, mutations, None)
.map(|_| ())
.map_err(refusal)
}
pub(super) fn create_ref(
db: &FeltDb,
namespace: &str,
name: &str,
target: StateId,
authority: String,
metadata: BTreeMap<String, Value>,
) -> Result<StateRef, RefError> {
let ref_k = ref_key(namespace, name);
let inc_k = incarnation_key(namespace, name);
let target_k = revision_key(&target);
let payload = payload_hash(&json!({
"target": target.as_hex(),
"authority": authority,
"metadata": metadata,
}));
for _ in 0..ATTEMPTS {
let incarnation_row = row(db, &inc_k)?;
if let Some(existing) = row(db, &ref_k)? {
let id = ref_transaction_id("create", namespace, name, incarnation_of(&existing), None);
if committed_payload(db, &id)?.as_deref() == Some(payload.as_str()) {
return decode(existing);
}
return Err(RefError::AlreadyExists);
}
if row(db, &target_k)?.is_none() {
return Err(RefError::UnknownTarget(target));
}
let incarnation = incarnation_in(incarnation_row.as_ref()) + 1;
let id = ref_transaction_id("create", namespace, name, incarnation, None);
let reference = StateRef {
namespace: namespace.to_string(),
name: name.to_string(),
target: target.clone(),
version: 1,
authority: authority.clone(),
updated_at_ms: current_time_ms(),
metadata: metadata.clone(),
};
let pins = [
unchanged(&ref_k, None),
unchanged(&inc_k, incarnation_row.as_ref()),
present(&target_k),
];
let mutations = [
put(&ref_k, ref_row(&reference, incarnation)),
put(&inc_k, json!({ "incarnation": incarnation })),
advance_generation(&id),
];
match commit(db, &id, Some(&payload), &pins, &mutations) {
Ok(()) => return Ok(reference),
Err(Refusal::Changed | Refusal::OtherRequest) => continue,
Err(Refusal::Failed(message)) => return Err(RefError::Storage(message)),
}
}
Err(RefError::Storage(contended("ref create")))
}
pub(super) fn compare_and_swap_ref(
db: &FeltDb,
namespace: &str,
name: &str,
expected_version: u64,
target: StateId,
authority: String,
metadata: Option<BTreeMap<String, Value>>,
) -> Result<StateRef, RefError> {
let ref_k = ref_key(namespace, name);
let inc_k = incarnation_key(namespace, name);
let target_k = revision_key(&target);
let payload = payload_hash(&json!({
"target": target.as_hex(),
"authority": authority,
"metadata": metadata,
}));
for _ in 0..ATTEMPTS {
let incarnation_row = row(db, &inc_k)?;
let Some(ref_read) = row(db, &ref_k)? else {
return Err(RefError::NotFound);
};
let incarnation = incarnation_of(&ref_read);
let current = decode(ref_read.clone())?;
let id = ref_transaction_id("cas", namespace, name, incarnation, Some(expected_version));
if current.version != expected_version {
if committed_payload(db, &id)?.as_deref() == Some(payload.as_str()) {
if current.version == expected_version + 1 && current.target == target {
return Ok(current);
}
return Ok(StateRef {
version: expected_version + 1,
target,
authority,
updated_at_ms: current_time_ms(),
metadata: metadata.unwrap_or(current.metadata),
..current
});
}
return Err(RefError::VersionConflict {
expected: expected_version,
current: current.version,
});
}
if row(db, &target_k)?.is_none() {
return Err(RefError::UnknownTarget(target));
}
let next = StateRef {
version: expected_version + 1,
target: target.clone(),
authority: authority.clone(),
updated_at_ms: current_time_ms(),
metadata: metadata.clone().unwrap_or_else(|| current.metadata.clone()),
..current
};
let pins = [
unchanged(&ref_k, Some(&ref_read)),
unchanged(&inc_k, incarnation_row.as_ref()),
present(&target_k),
];
let mutations = [put(&ref_k, ref_row(&next, incarnation)), advance_generation(&id)];
match commit(db, &id, Some(&payload), &pins, &mutations) {
Ok(()) => return Ok(next),
Err(Refusal::Changed | Refusal::OtherRequest) => continue,
Err(Refusal::Failed(message)) => return Err(RefError::Storage(message)),
}
}
Err(RefError::Storage(contended("ref compare-and-swap")))
}
pub(super) fn delete_ref(
db: &FeltDb,
namespace: &str,
name: &str,
expected_version: Option<u64>,
) -> Result<(), RefError> {
let ref_k = ref_key(namespace, name);
let inc_k = incarnation_key(namespace, name);
let payload = payload_hash(&json!({ "expected_version": expected_version }));
for _ in 0..ATTEMPTS {
let incarnation_row = row(db, &inc_k)?;
let Some(ref_read) = row(db, &ref_k)? else {
let id = ref_transaction_id(
"delete",
namespace,
name,
incarnation_in(incarnation_row.as_ref()),
None,
);
if committed_payload(db, &id)?.as_deref() == Some(payload.as_str()) {
return Ok(());
}
return Err(RefError::NotFound);
};
let incarnation = incarnation_of(&ref_read);
let current = decode(ref_read.clone())?;
if let Some(expected) = expected_version {
if current.version != expected {
return Err(RefError::VersionConflict {
expected,
current: current.version,
});
}
}
let id = ref_transaction_id("delete", namespace, name, incarnation, None);
let pins = [
unchanged(&ref_k, Some(&ref_read)),
unchanged(&inc_k, incarnation_row.as_ref()),
];
let mutations = [
remove(&ref_k),
put(&inc_k, json!({ "incarnation": incarnation })),
advance_generation(&id),
];
match commit(db, &id, Some(&payload), &pins, &mutations) {
Ok(()) => return Ok(()),
Err(Refusal::Changed | Refusal::OtherRequest) => continue,
Err(Refusal::Failed(message)) => return Err(RefError::Storage(message)),
}
}
Err(RefError::Storage(contended("ref delete")))
}
pub(super) fn commit_revision(
db: &FeltDb,
revision: StateRevision,
) -> Result<StateRevision, RevisionError> {
let fail = |error: &dyn std::fmt::Display| RevisionError::Storage(error.to_string());
let object = StateObject::new(revision.content.clone());
let rev_k = revision_key(&revision.id);
let obj_k = object_key(&object.id);
let parent_k = revision.parent_id.as_ref().map(revision_key);
for _ in 0..ATTEMPTS {
if let Some(existing) = db.get::<StateRevision>(&rev_k).map_err(|e| fail(&e))? {
let same = existing.resource == revision.resource
&& existing.content_id == revision.content_id
&& existing.parent_id == revision.parent_id
&& existing.sequence == revision.sequence;
return if same {
Ok(existing)
} else {
Err(RevisionError::AlreadyCommitted(revision.id.clone()))
};
}
let mut pins = vec![unchanged(&rev_k, None)];
if let (Some(parent), Some(parent_k)) = (&revision.parent_id, &parent_k) {
if db.get::<Value>(parent_k).map_err(|e| fail(&e))?.is_none() {
return Err(RevisionError::UnknownParent(parent.clone()));
}
pins.push(present(parent_k));
}
let id = format!("state-revision:{}:{}", revision.id.as_hex(), unique_suffix());
let mut mutations = Vec::with_capacity(3);
match db.get::<StateObject>(&obj_k).map_err(|e| fail(&e))? {
Some(existing) if !existing.verify_integrity() => {
return Err(RevisionError::Storage(
"Stored state object failed integrity verification".to_string(),
));
}
Some(_) => pins.push(present(&obj_k)),
None => {
pins.push(unchanged(&obj_k, None));
mutations.push(put(&obj_k, serde_json::to_value(&object).map_err(|e| fail(&e))?));
}
}
mutations.push(put(&rev_k, serde_json::to_value(&revision).map_err(|e| fail(&e))?));
mutations.push(advance_generation(&id));
match commit(db, &id, None, &pins, &mutations) {
Ok(()) => return Ok(revision),
Err(Refusal::Changed | Refusal::OtherRequest) => continue,
Err(Refusal::Failed(message)) => return Err(RevisionError::Storage(message)),
}
}
Err(RevisionError::Storage(contended("revision commit")))
}
pub(super) fn collect(
store: &StateStore,
db: &FeltDb,
now_ms: u64,
) -> Result<GarbageCollectionReport, GarbageCollectionError> {
let fail = |error: &dyn std::fmt::Display| GarbageCollectionError::Storage(error.to_string());
for _ in 0..ATTEMPTS {
let generation = db.get::<Value>(GENERATION_KEY).map_err(|e| fail(&e))?;
let report = store.plan_collection(now_ms, &db.revision_head_ids());
if report.collected_revisions.is_empty() && report.collected_objects.is_empty() {
return Ok(report);
}
let pins = [unchanged(GENERATION_KEY, generation.as_ref())];
let mutations: Vec<AtomicMutation> = report
.collected_revisions
.iter()
.map(|id| remove(&revision_key(id)))
.chain(report.collected_objects.iter().map(|id| remove(&object_key(id))))
.collect();
let id = format!("state-gc:{}", unique_suffix());
match commit(db, &id, None, &pins, &mutations) {
Ok(()) => return Ok(report),
Err(Refusal::Changed | Refusal::OtherRequest) => continue,
Err(Refusal::Failed(message)) => return Err(GarbageCollectionError::Storage(message)),
}
}
Err(GarbageCollectionError::Storage(contended("garbage collection")))
}
}
fn collect_referenced_objects(revision: &StateRevision) -> BTreeSet<StateId> {
let mut ids = BTreeSet::from([revision.content_id.clone()]);
if let Some(values) = revision
.metadata
.get("object_refs")
.and_then(Value::as_array)
{
for value in values {
if let Some(hex) = value.as_str().filter(|value| looks_like_state_id(value)) {
ids.insert(StateId::from_hex(hex.to_string()));
}
}
}
ids
}
fn retention_key(resource: &str) -> String {
format!("_retention:{resource}")
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RetentionPolicy {
pub keep_last: Option<usize>,
}
impl RetentionPolicy {
pub const fn unbounded() -> Self {
RetentionPolicy { keep_last: None }
}
pub fn keep_last(count: usize) -> Self {
RetentionPolicy {
keep_last: Some(count.max(1)),
}
}
}
impl Default for RetentionPolicy {
fn default() -> Self {
Self::unbounded()
}
}
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
struct RetentionState {
policy: RetentionPolicy,
horizon: u64,
}
pub fn revisions_to_expire(
history: &[(StateId, u64)],
keep_last: Option<usize>,
) -> (Vec<StateId>, Option<u64>) {
let Some(keep) = keep_last else {
return (Vec::new(), None);
};
let keep = keep.max(1);
if history.len() <= keep {
return (Vec::new(), None);
}
let split = history.len() - keep;
let expire = history[..split].iter().map(|(id, _)| id.clone()).collect();
let horizon = history[split].1;
(expire, Some(horizon))
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RevisionError {
UnknownParent(StateId),
AlreadyCommitted(StateId),
ForeignParent { expected: String, found: String },
ResourceAlreadyBegun { resource: String, head: StateId },
IntegrityFailed,
Storage(String),
}
impl std::fmt::Display for RevisionError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
RevisionError::UnknownParent(id) => {
write!(f, "parent {id} is not a revision this store holds")
}
RevisionError::AlreadyCommitted(id) => {
write!(
f,
"revision {id} is already committed and cannot be rewritten"
)
}
RevisionError::ForeignParent { expected, found } => {
write!(f, "parent belongs to resource {found}, not {expected}")
}
RevisionError::ResourceAlreadyBegun { resource, head } => {
write!(f, "resource {resource} already begins at revision {head}")
}
RevisionError::IntegrityFailed => write!(f, "state integrity verification failed"),
RevisionError::Storage(message) => write!(f, "{message}"),
}
}
}
impl std::error::Error for RevisionError {}
#[derive(Clone, Debug)]
pub enum ParentLookup {
Root,
Revision(Box<StateRevision>),
Expired(StateId),
Missing(StateId),
}
impl ParentLookup {
pub fn id(&self) -> Option<&StateId> {
match self {
ParentLookup::Root => None,
ParentLookup::Revision(revision) => Some(&revision.id),
ParentLookup::Expired(id) | ParentLookup::Missing(id) => Some(id),
}
}
}
impl PartialEq for ParentLookup {
fn eq(&self, other: &Self) -> bool {
std::mem::discriminant(self) == std::mem::discriminant(other) && self.id() == other.id()
}
}
impl StateStore {
pub fn new_volatile() -> Self {
StateStore {
backing: Backing::Volatile(Arc::new(Mutex::new(VolatileState::default()))),
mutation_lock: Arc::new(Mutex::new(())),
}
}
pub fn with_feltdb(feltdb: Arc<crate::FeltDb>) -> Result<Self, String> {
Ok(StateStore {
backing: Backing::Durable(feltdb),
mutation_lock: Arc::new(Mutex::new(())),
})
}
pub fn create(
&self,
resource: &str,
content: String,
authority: String,
) -> Result<StateRevision, RevisionError> {
if let Some(head) = self.head_of(resource) {
return Err(RevisionError::ResourceAlreadyBegun {
resource: resource.to_string(),
head: head.id,
});
}
self.mint(resource, content, None, authority)
}
pub fn commit(
&self,
content: String,
parent: &StateRevision,
authority: String,
) -> Result<StateRevision, RevisionError> {
let resource = parent.resource.clone();
self.mint(&resource, content, Some(parent), authority)
}
pub fn mint(
&self,
resource: &str,
content: String,
parent: Option<&StateRevision>,
authority: String,
) -> Result<StateRevision, RevisionError> {
if let Some(parent) = parent {
if parent.resource != resource {
return Err(RevisionError::ForeignParent {
expected: resource.to_string(),
found: parent.resource.clone(),
});
}
if !self.exists(&parent.id) {
return Err(RevisionError::UnknownParent(parent.id.clone()));
}
}
let sequence = match parent {
Some(parent) => parent.sequence + 1,
None => 0,
};
let revision = StateRevision::at(
resource.to_string(),
content,
parent.map(|parent| parent.id.clone()),
sequence,
authority,
);
let committed = self.commit_revision(revision)?;
self.apply_retention(resource)?;
Ok(committed)
}
fn commit_revision(&self, revision: StateRevision) -> Result<StateRevision, RevisionError> {
if !revision.verify_integrity() {
return Err(RevisionError::IntegrityFailed);
}
if let Some(existing) = self.get(&revision.id) {
let same = existing.resource == revision.resource
&& existing.content_id == revision.content_id
&& existing.parent_id == revision.parent_id
&& existing.sequence == revision.sequence;
return if same {
Ok(existing)
} else {
Err(RevisionError::AlreadyCommitted(revision.id.clone()))
};
}
match &self.backing {
Backing::Volatile(state) => {
let mut state = state.lock().unwrap();
state
.objects
.entry(revision.content_id.clone())
.or_insert_with(|| StateObject::new(revision.content.clone()));
state.revisions.insert(revision.id.clone(), revision.clone());
}
Backing::Durable(feltdb) => {
return durable::commit_revision(feltdb, revision);
}
}
Ok(revision)
}
pub fn get(&self, id: &StateId) -> Option<StateRevision> {
match &self.backing {
Backing::Volatile(state) => state.lock().unwrap().revisions.get(id).cloned(),
Backing::Durable(feltdb) => feltdb
.get::<StateRevision>(&revision_key(id))
.ok()
.flatten(),
}
}
pub fn exists(&self, id: &StateId) -> bool {
self.get(id).is_some()
}
pub fn metadata(&self, id: &StateId) -> Option<BTreeMap<String, Value>> {
self.get(id).map(|revision| revision.metadata)
}
pub fn parent_of(&self, id: &StateId) -> Option<ParentLookup> {
let revision = self.get(id)?;
let Some(parent_id) = revision.parent_id else {
return Some(ParentLookup::Root);
};
if let Some(parent) = self.get(&parent_id) {
return Some(ParentLookup::Revision(Box::new(parent)));
}
let horizon = self.retention_state(&revision.resource).horizon;
Some(if revision.sequence <= horizon {
ParentLookup::Expired(parent_id)
} else {
ParentLookup::Missing(parent_id)
})
}
pub fn parent(&self, id: &StateId) -> Option<StateRevision> {
match self.parent_of(id)? {
ParentLookup::Revision(parent) => Some(*parent),
_ => None,
}
}
pub fn history_of(&self, resource: &str) -> Vec<StateRevision> {
let mut history: Vec<StateRevision> = self
.all_revisions()
.into_iter()
.filter(|revision| revision.resource == resource)
.collect();
history.sort_by_key(|revision| revision.sequence);
history
}
pub fn head_of(&self, resource: &str) -> Option<StateRevision> {
self.history_of(resource).into_iter().max_by(|one, two| {
one.sequence
.cmp(&two.sequence)
.then_with(|| one.id.as_hex().cmp(two.id.as_hex()))
})
}
pub fn resources(&self) -> Vec<String> {
let mut names: Vec<String> = self
.all_revisions()
.into_iter()
.map(|revision| revision.resource)
.collect();
names.sort();
names.dedup();
names
}
pub fn create_ref(
&self,
namespace: &str,
name: &str,
target: StateId,
authority: String,
metadata: BTreeMap<String, Value>,
) -> Result<StateRef, RefError> {
if !valid_ref_component(namespace) || !valid_ref_component(name) {
return Err(RefError::InvalidIdentifier);
}
match &self.backing {
Backing::Durable(feltdb) => {
durable::create_ref(feltdb, namespace, name, target, authority, metadata)
}
Backing::Volatile(_) => {
let _guard = self.mutation_lock.lock().unwrap();
if !self.exists(&target) {
return Err(RefError::UnknownTarget(target));
}
if self.get_ref(namespace, name).is_some() {
return Err(RefError::AlreadyExists);
}
let reference = StateRef {
namespace: namespace.to_string(),
name: name.to_string(),
target,
version: 1,
authority,
updated_at_ms: current_time_ms(),
metadata,
};
self.write_volatile_ref(&reference);
Ok(reference)
}
}
}
pub fn get_ref(&self, namespace: &str, name: &str) -> Option<StateRef> {
if !valid_ref_component(namespace) || !valid_ref_component(name) {
return None;
}
match &self.backing {
Backing::Volatile(state) => state
.lock()
.unwrap()
.refs
.get(&(namespace.to_string(), name.to_string()))
.cloned(),
Backing::Durable(feltdb) => feltdb
.get::<StateRef>(&ref_key(namespace, name))
.ok()
.flatten(),
}
}
pub fn move_ref(
&self,
namespace: &str,
name: &str,
target: StateId,
authority: String,
) -> Result<StateRef, RefError> {
let current = self.get_ref(namespace, name).ok_or(RefError::NotFound)?;
self.compare_and_swap_ref(namespace, name, current.version, target, authority, None)
}
pub fn compare_and_swap_ref(
&self,
namespace: &str,
name: &str,
expected_version: u64,
target: StateId,
authority: String,
metadata: Option<BTreeMap<String, Value>>,
) -> Result<StateRef, RefError> {
if !valid_ref_component(namespace) || !valid_ref_component(name) {
return Err(RefError::InvalidIdentifier);
}
match &self.backing {
Backing::Durable(feltdb) => durable::compare_and_swap_ref(
feltdb,
namespace,
name,
expected_version,
target,
authority,
metadata,
),
Backing::Volatile(_) => {
let _guard = self.mutation_lock.lock().unwrap();
let mut current = self.get_ref(namespace, name).ok_or(RefError::NotFound)?;
if current.version != expected_version {
return Err(RefError::VersionConflict {
expected: expected_version,
current: current.version,
});
}
if !self.exists(&target) {
return Err(RefError::UnknownTarget(target));
}
current.target = target;
current.version += 1;
current.authority = authority;
current.updated_at_ms = current_time_ms();
if let Some(metadata) = metadata {
current.metadata = metadata;
}
self.write_volatile_ref(¤t);
Ok(current)
}
}
}
pub fn delete_ref(
&self,
namespace: &str,
name: &str,
expected_version: Option<u64>,
) -> Result<(), RefError> {
if !valid_ref_component(namespace) || !valid_ref_component(name) {
return Err(RefError::InvalidIdentifier);
}
match &self.backing {
Backing::Durable(feltdb) => {
durable::delete_ref(feltdb, namespace, name, expected_version)
}
Backing::Volatile(state) => {
let _guard = self.mutation_lock.lock().unwrap();
let current = self.get_ref(namespace, name).ok_or(RefError::NotFound)?;
if let Some(expected) = expected_version {
if current.version != expected {
return Err(RefError::VersionConflict {
expected,
current: current.version,
});
}
}
state
.lock()
.unwrap()
.refs
.remove(&(namespace.to_string(), name.to_string()));
Ok(())
}
}
}
pub fn list_refs(&self, namespace: Option<&str>) -> Vec<StateRef> {
let mut refs: Vec<StateRef> = match &self.backing {
Backing::Volatile(state) => state.lock().unwrap().refs.values().cloned().collect(),
Backing::Durable(feltdb) => feltdb
.list_collection("state")
.unwrap_or_default()
.into_iter()
.filter(|row| row.key.starts_with("state:ref:") && !row.deleted)
.filter_map(|row| serde_json::from_value::<StateRef>(row.value).ok())
.collect(),
};
if let Some(namespace) = namespace {
refs.retain(|reference| reference.namespace == namespace);
}
refs.sort_by(|one, two| {
one.namespace
.cmp(&two.namespace)
.then_with(|| one.name.cmp(&two.name))
});
refs
}
pub fn materialize(
&self,
revision_id: &StateId,
projection: &MaterializationProjection,
) -> Result<Value, MaterializationError> {
let revision = self
.get(revision_id)
.ok_or_else(|| MaterializationError::RevisionNotFound(revision_id.clone()))?;
let source: Value = serde_json::from_str(&revision.content)
.map_err(|error| MaterializationError::ContentUnparseable(error.to_string()))?;
if projection.is_full() {
return Ok(source);
}
let mut paths = projection.paths.clone();
paths.sort();
paths.dedup();
for pair in paths.windows(2) {
if paths_overlap(&pair[0], &pair[1]) {
return Err(MaterializationError::NonDeterministicProjection);
}
}
let mut projected = Value::Object(serde_json::Map::new());
for path in paths {
let value = resolve_path(&source, &path)
.cloned()
.ok_or_else(|| MaterializationError::MissingPath(path.clone()))?;
assign_projection(&mut projected, &path, value)?;
}
Ok(projected)
}
pub fn collect_unreachable(&self) -> Result<GarbageCollectionReport, GarbageCollectionError> {
self.collect_unreachable_at(current_time_ms())
}
pub(crate) fn collect_unreachable_at(
&self,
now_ms: u64,
) -> Result<GarbageCollectionReport, GarbageCollectionError> {
match &self.backing {
Backing::Volatile(state) => {
let _guard = self.mutation_lock.lock().unwrap();
let report = self.plan_collection(now_ms, &[]);
let mut state = state.lock().unwrap();
for id in &report.collected_revisions {
state.revisions.remove(id);
}
for id in &report.collected_objects {
state.objects.remove(id);
}
Ok(report)
}
Backing::Durable(feltdb) => durable::collect(self, feltdb, now_ms),
}
}
fn plan_collection(&self, now_ms: u64, extra_roots: &[StateId]) -> GarbageCollectionReport {
let roots = self.list_refs(None);
let revisions = self.all_revisions();
let by_id: HashMap<StateId, &StateRevision> =
revisions.iter().map(|revision| (revision.id.clone(), revision)).collect();
let mut latest: HashMap<&str, u64> = HashMap::new();
for revision in &revisions {
let entry = latest.entry(revision.resource.as_str()).or_insert(revision.sequence);
if revision.sequence > *entry {
*entry = revision.sequence;
}
}
let mut pending: Vec<StateId> = roots.iter().map(|reference| reference.target.clone()).collect();
pending.extend(extra_roots.iter().cloned());
for revision in &revisions {
let is_latest = latest.get(revision.resource.as_str()) == Some(&revision.sequence);
let is_young = now_ms.saturating_sub(revision.timestamp_ms) < GC_GRACE_MS;
if is_latest || is_young {
pending.push(revision.id.clone());
}
}
let mut reachable_revisions = BTreeSet::new();
let mut reachable_objects = BTreeSet::new();
while let Some(id) = pending.pop() {
let Some(revision) = by_id.get(&id) else {
continue;
};
if !reachable_revisions.insert(id.clone()) {
continue;
}
if let Some(parent_id) = revision.parent_id.clone() {
pending.push(parent_id);
}
reachable_objects.extend(collect_referenced_objects(revision));
}
let mut collected_revisions: Vec<StateId> = revisions
.iter()
.map(|revision| revision.id.clone())
.filter(|id| !reachable_revisions.contains(id))
.collect();
collected_revisions.sort();
let mut collected_objects: Vec<StateId> = self
.all_objects()
.into_iter()
.map(|object| object.id)
.filter(|id| !reachable_objects.contains(id))
.collect();
collected_objects.sort();
GarbageCollectionReport {
roots,
reachable_revisions: reachable_revisions.into_iter().collect(),
reachable_objects: reachable_objects.into_iter().collect(),
collected_revisions,
collected_objects,
}
}
fn write_volatile_ref(&self, reference: &StateRef) {
if let Backing::Volatile(state) = &self.backing {
state.lock().unwrap().refs.insert(
(reference.namespace.clone(), reference.name.clone()),
reference.clone(),
);
}
}
fn all_revisions(&self) -> Vec<StateRevision> {
match &self.backing {
Backing::Volatile(state) => state.lock().unwrap().revisions.values().cloned().collect(),
Backing::Durable(feltdb) => feltdb
.list_collection("state")
.unwrap_or_default()
.into_iter()
.filter(|row| row.key.starts_with("state:revision:") && !row.deleted)
.filter_map(|row| serde_json::from_value::<StateRevision>(row.value).ok())
.collect(),
}
}
fn all_objects(&self) -> Vec<StateObject> {
match &self.backing {
Backing::Volatile(state) => state.lock().unwrap().objects.values().cloned().collect(),
Backing::Durable(feltdb) => feltdb
.list_collection("state")
.unwrap_or_default()
.into_iter()
.filter(|row| row.key.starts_with("state:object:") && !row.deleted)
.filter_map(|row| serde_json::from_value::<StateObject>(row.value).ok())
.collect(),
}
}
pub fn history_digest(&self) -> String {
let mut entries: Vec<String> = self
.all_revisions()
.into_iter()
.map(|revision| {
format!(
"{}|{}|{}|{}|{}|{}",
revision.resource,
revision.id.as_hex(),
revision
.parent_id
.as_ref()
.map(StateId::as_hex)
.unwrap_or("-"),
revision.sequence,
revision.content_id.as_hex(),
revision.authority,
)
})
.collect();
for resource in self.resources() {
let state = self.retention_state(&resource);
entries.push(format!(
"retention|{}|{:?}|{}",
resource, state.policy.keep_last, state.horizon
));
}
entries.sort();
let mut hasher = Sha256::new();
hasher.update(b"feltdb.history.v1");
for entry in entries {
hasher.update(entry.as_bytes());
hasher.update([0u8]);
}
format!("{:x}", hasher.finalize())
}
pub fn retention_policy(&self, resource: &str) -> RetentionPolicy {
self.retention_state(resource).policy
}
pub fn retention_horizon(&self, resource: &str) -> u64 {
self.retention_state(resource).horizon
}
pub fn set_retention_policy(
&self,
resource: &str,
policy: RetentionPolicy,
) -> Result<usize, RevisionError> {
let mut state = self.retention_state(resource);
state.policy = policy;
self.write_retention_state(resource, &state)?;
self.apply_retention(resource)
}
pub fn apply_retention(&self, resource: &str) -> Result<usize, RevisionError> {
let mut state = self.retention_state(resource);
if state.policy.keep_last.is_none() {
return Ok(0);
}
let history: Vec<(StateId, u64)> = self
.history_of(resource)
.into_iter()
.map(|revision| (revision.id, revision.sequence))
.collect();
let (expire, horizon) = revisions_to_expire(&history, state.policy.keep_last);
if expire.is_empty() {
return Ok(0);
}
for id in &expire {
self.remove(id)?;
}
if let Some(horizon) = horizon {
state.horizon = horizon;
}
self.write_retention_state(resource, &state)?;
Ok(expire.len())
}
fn remove(&self, id: &StateId) -> Result<(), RevisionError> {
match &self.backing {
Backing::Volatile(state) => {
state.lock().unwrap().revisions.remove(id);
}
Backing::Durable(feltdb) => {
feltdb.delete(&revision_key(id)).map_err(|error| {
RevisionError::Storage(format!("Failed to expire revision: {error}"))
})?;
}
}
Ok(())
}
fn retention_state(&self, resource: &str) -> RetentionState {
match &self.backing {
Backing::Volatile(state) => state
.lock()
.unwrap()
.retention
.get(resource)
.cloned()
.unwrap_or_default(),
Backing::Durable(feltdb) => feltdb
.get::<RetentionState>(&retention_key(resource))
.ok()
.flatten()
.unwrap_or_default(),
}
}
fn write_retention_state(
&self,
resource: &str,
state: &RetentionState,
) -> Result<(), RevisionError> {
match &self.backing {
Backing::Volatile(volatile) => {
volatile
.lock()
.unwrap()
.retention
.insert(resource.to_string(), state.clone());
}
Backing::Durable(feltdb) => {
feltdb
.insert(&retention_key(resource), state.clone())
.map_err(|error| {
RevisionError::Storage(format!(
"Failed to persist retention state: {error}"
))
})?;
}
}
Ok(())
}
}
impl Clone for StateStore {
fn clone(&self) -> Self {
StateStore {
backing: self.backing.clone(),
mutation_lock: self.mutation_lock.clone(),
}
}
}
impl Default for StateStore {
fn default() -> Self {
Self::new_volatile()
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn forked(store: &StateStore) -> (StateRevision, StateRevision, StateRevision, StateRevision) {
let a = store.create("doc:gc", json!({"n": "a"}).to_string(), "t".into()).unwrap();
let b = store.commit(json!({"n": "b"}).to_string(), &a, "t".into()).unwrap();
let c = store.commit(json!({"n": "c"}).to_string(), &b, "t".into()).unwrap();
let f = store.commit(json!({"n": "f"}).to_string(), &a, "t".into()).unwrap();
(a, b, c, f)
}
#[test]
fn gc_grace_boundary_is_exclusive() {
let store = StateStore::new_volatile();
let (_, _, _, f) = forked(&store);
let inside = store.collect_unreachable_at(f.timestamp_ms + GC_GRACE_MS - 1).unwrap();
assert!(inside.collected_revisions.is_empty(), "younger than the grace: a root");
let outside = store.collect_unreachable_at(f.timestamp_ms + GC_GRACE_MS).unwrap();
assert_eq!(outside.collected_revisions, vec![f.id.clone()]);
assert!(!store.exists(&f.id));
}
#[test]
fn gc_keeps_latest_revisions_refs_and_all_their_ancestors() {
for store in [StateStore::new_volatile(), durable_store().1] {
let (a, b, c, f) = forked(&store);
let g = store.commit(json!({"n": "g"}).to_string(), &f, "t".into()).unwrap();
let far = c.timestamp_ms + 365 * 24 * 60 * 60 * 1000;
let report = store.collect_unreachable_at(far).unwrap();
assert!(report.collected_revisions.is_empty(), "every revision is a root or an ancestor");
let d = store.commit(json!({"n": "d"}).to_string(), &c, "t".into()).unwrap();
store
.create_ref("candidates", "g", g.id.clone(), "t".into(), BTreeMap::new())
.unwrap();
assert!(store.collect_unreachable_at(far).unwrap().collected_revisions.is_empty());
store.delete_ref("candidates", "g", None).unwrap();
let mut collected = store.collect_unreachable_at(far).unwrap().collected_revisions;
collected.sort();
let mut expected = vec![f.id.clone(), g.id.clone()];
expected.sort();
assert_eq!(collected, expected, "only the unreferenced stale fork");
for kept in [&a, &b, &c, &d] {
assert!(store.exists(&kept.id));
}
let objects: BTreeSet<StateId> = store.all_objects().into_iter().map(|o| o.id).collect();
for kept in [&a, &b, &c, &d] {
assert!(objects.contains(&kept.content_id), "object of a kept revision");
}
for gone in [&f, &g] {
assert!(!objects.contains(&gone.content_id), "object of a collected revision");
}
}
}
fn durable_store() -> (tempfile::TempDir, StateStore, Arc<crate::FeltDb>) {
let dir = tempfile::TempDir::new().unwrap();
let db = Arc::new(crate::FeltDb::open(dir.path().join("gc.log")).unwrap());
let store = StateStore::with_feltdb(db.clone()).unwrap();
(dir, store, db)
}
#[test]
fn gc_keeps_the_write_path_head_even_when_a_fork_is_longer() {
let (_dir, store, db) = durable_store();
db.insert("tasks:1", json!({"n": 0})).unwrap();
db.update("tasks:1", json!({"n": 1})).unwrap();
let resource = store.resources().into_iter().find(|r| r.contains("tasks")).unwrap();
let history = store.history_of(&resource);
assert_eq!(history.len(), 2, "the write path minted two revisions");
let (w0, w1) = (history[0].clone(), history[1].clone());
let x = store.commit(json!({"x": 1}).to_string(), &w0, "t".into()).unwrap();
let x2 = store.commit(json!({"x": 2}).to_string(), &x, "t".into()).unwrap();
assert_eq!(store.head_of(&resource).unwrap().id, x2.id);
assert!(db.revision_head_ids().contains(&w1.id), "the write path still mints from w1");
let far = x2.timestamp_ms + 365 * 24 * 60 * 60 * 1000;
let report = store.collect_unreachable_at(far).unwrap();
assert!(!report.collected_revisions.contains(&w1.id), "w1 is the write path's head");
assert!(store.exists(&w1.id));
db.update("tasks:1", json!({"n": 2})).unwrap();
let w2 = store
.history_of(&resource)
.into_iter()
.find(|r| r.parent_id.as_ref() == Some(&w1.id))
.expect("the next write minted a child of w1");
assert_eq!(store.parent_of(&w2.id), Some(ParentLookup::Revision(Box::new(w1))));
}
#[test]
fn internal_rows_do_not_appear_as_history_or_refs() {
let (_dir, store, _db) = durable_store();
let (a, _, _, _) = forked(&store);
store.create_ref("heads", "main", a.id.clone(), "t".into(), BTreeMap::new()).unwrap();
store.delete_ref("heads", "main", Some(1)).unwrap();
store.create_ref("heads", "main", a.id.clone(), "t".into(), BTreeMap::new()).unwrap();
assert_eq!(store.list_refs(None).len(), 1);
assert_eq!(store.resources(), vec!["doc:gc".to_string()]);
assert_eq!(store.history_of("doc:gc").len(), 4);
}
#[test]
fn test_state_id_deterministic() {
let json = r#"{"name":"Alice","age":30}"#;
let id1 = StateId::compute(json);
let id2 = StateId::compute(json);
assert_eq!(id1, id2);
}
#[test]
fn test_state_id_representation_sensitive() {
let id_int = StateId::compute("1");
let id_float = StateId::compute("1.0");
let id_string = StateId::compute(r#""1""#);
assert_ne!(id_int, id_float);
assert_ne!(id_int, id_string);
assert_ne!(id_float, id_string);
}
#[test]
fn test_state_revision_initial() {
let rev = StateRevision::initial(
"docs:1".to_string(),
r#"{"key":"value"}"#.to_string(),
"test-authority".to_string(),
);
assert!(rev.parent_id.is_none());
assert_eq!(rev.authority, "test-authority");
assert!(rev.verify_integrity());
}
#[test]
fn test_state_revision_child() {
let parent = StateRevision::initial(
"docs:1".to_string(),
r#"{"key":"value"}"#.to_string(),
"test-authority".to_string(),
);
let child = StateRevision::child(
r#"{"key":"updated"}"#.to_string(),
&parent,
"test-authority".to_string(),
);
assert_eq!(child.parent_id, Some(parent.id.clone()));
assert!(child.verify_integrity());
}
#[test]
fn test_topology_ancestry() {
let mut topo = StateTopology::new();
let rev1 = StateRevision::initial(
"docs:1".to_string(),
r#"{"v":1}"#.to_string(),
"auth".to_string(),
);
let rev2 = StateRevision::child(r#"{"v":2}"#.to_string(), &rev1, "auth".to_string());
topo.add_revision(rev1.clone());
topo.add_revision(rev2.clone());
assert!(topo.is_ancestor(&rev1.id, &rev2.id));
assert!(!topo.is_ancestor(&rev2.id, &rev1.id));
}
#[test]
fn test_semantic_diff_simple() {
let old = json!({"a": 1, "b": 2});
let new = json!({"a": 1, "b": 3});
let diff = SemanticDiff::compute(&old, &new);
assert_eq!(diff.changes.len(), 1);
assert_eq!(diff.changes[0].kind, ChangeKind::Changed);
assert_eq!(diff.changes[0].old_value, Some(json!(2)));
assert_eq!(diff.changes[0].new_value, Some(json!(3)));
}
#[test]
fn path_relation_is_structural() {
let key = |name: &str| PathComponent::Key(name.to_string());
assert_eq!(path_relation(&[key("a")], &[key("a")]), PathRelation::Same);
assert_eq!(
path_relation(&[key("a")], &[key("a"), key("b")]),
PathRelation::Ancestor
);
assert_eq!(
path_relation(&[key("a"), key("b")], &[key("a")]),
PathRelation::Descendant
);
assert_eq!(
path_relation(&[key("a"), key("b")], &[key("a"), key("c")]),
PathRelation::Disjoint
);
assert_eq!(
path_relation(&[key("a"), key("b")], &[key("a"), key("bc")]),
PathRelation::Disjoint
);
assert_eq!(
path_relation(&[key("0")], &[PathComponent::Index(0)]),
PathRelation::Disjoint
);
assert_eq!(path_relation(&[], &[key("a")]), PathRelation::Ancestor);
assert_eq!(path_relation(&[], &[]), PathRelation::Same);
assert!(paths_overlap(&[key("a")], &[key("a"), key("b")]));
assert!(paths_overlap(&[key("a"), key("b")], &[key("a")]));
assert!(!paths_overlap(
&[key("a"), key("b")],
&[key("a"), key("bc")]
));
assert!(!paths_overlap(&[key("a")], &[key("b")]));
}
#[test]
fn resolve_path_reads_objects_and_arrays() {
let value = json!({"o": {"a": [10, {"b": 1}]}, "n": null});
let key = |name: &str| PathComponent::Key(name.to_string());
assert_eq!(resolve_path(&value, &[]), Some(&value));
assert_eq!(resolve_path(&value, &[key("n")]), Some(&Value::Null));
assert_eq!(
resolve_path(&value, &[key("o"), key("a"), PathComponent::Index(0)]),
Some(&json!(10))
);
assert_eq!(
resolve_path(
&value,
&[key("o"), key("a"), PathComponent::Index(1), key("b")]
),
Some(&json!(1))
);
assert_eq!(resolve_path(&value, &[key("missing")]), None);
assert_eq!(resolve_path(&value, &[PathComponent::Index(0)]), None);
assert_eq!(
resolve_path(&value, &[key("o"), key("a"), PathComponent::Index(9)]),
None
);
assert_eq!(resolve_path(&value, &[key("o"), key("a"), key("0")]), None);
}
#[test]
fn overlapping_changes_are_never_independent() {
let base = StateRevision::initial(
"docs:1".to_string(),
r#"{"o":{"a":1}}"#.to_string(),
"auth".to_string(),
);
let left = StateRevision::child(r#"{"o":7}"#.to_string(), &base, "auth".to_string());
let right = StateRevision::child(r#"{"o":{"a":2}}"#.to_string(), &base, "auth".to_string());
let classification = ConflictClassification::classify(&base, &left, &right);
assert_eq!(classification.overall, ConflictClass::Conflict);
assert!(classification
.path_conflicts
.iter()
.all(|entry| entry.classification == ConflictClass::Conflict));
}
#[test]
fn test_conflict_classification_independent() {
let base = StateRevision::initial(
"docs:1".to_string(),
r#"{"a":1,"b":1}"#.to_string(),
"auth".to_string(),
);
let left = StateRevision::child(r#"{"a":2,"b":1}"#.to_string(), &base, "auth".to_string());
let right = StateRevision::child(r#"{"a":1,"b":2}"#.to_string(), &base, "auth".to_string());
let classification = ConflictClassification::classify(&base, &left, &right);
assert_eq!(classification.overall, ConflictClass::Independent);
}
#[test]
fn test_state_store_create_and_get() {
let store = StateStore::new_volatile();
let rev = store
.create(
"docs:1",
r#"{"data":"initial"}"#.to_string(),
"auth".to_string(),
)
.unwrap();
let stored = store.get(&rev.id).unwrap();
assert_eq!(stored.id, rev.id);
assert_eq!(stored.content, rev.content);
}
#[test]
fn test_state_store_commit() {
let store = StateStore::new_volatile();
let rev1 = store
.create("docs:1", r#"{"v":1}"#.to_string(), "auth".to_string())
.unwrap();
let rev2 = store
.commit(r#"{"v":2}"#.to_string(), &rev1, "auth".to_string())
.unwrap();
assert_eq!(rev2.parent_id, Some(rev1.id.clone()));
assert_eq!(store.get(&rev2.id).unwrap().id, rev2.id);
assert_eq!(store.get(&rev1.id).unwrap().id, rev1.id);
assert_eq!(store.parent(&rev2.id).unwrap().id, rev1.id);
}
#[test]
fn test_reconciliation_plan_validation() {
let plan = ReconciliationPlan::new(
StateId::from_hex("aaa".to_string()),
StateId::from_hex("bbb".to_string()),
StateId::from_hex("ccc".to_string()),
true,
);
assert!(plan.validate());
}
#[test]
fn test_recovery_root_state() {
use std::sync::Arc;
let db_path = std::env::temp_dir().join("test_recovery_root.log");
let _ = std::fs::remove_file(&db_path);
let initial_id;
{
let db = crate::FeltDb::open(&db_path).expect("open db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let initial = store
.create(
"docs:1",
r#"{"data":"root"}"#.to_string(),
"auth".to_string(),
)
.expect("create initial");
assert_eq!(store.get(&initial.id).unwrap().id, initial.id);
initial_id = initial.id;
}
{
let db = crate::FeltDb::open(&db_path).expect("open db again");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store over same log");
let root = store.get(&initial_id).expect("the root survives restart");
assert_eq!(root.content, r#"{"data":"root"}"#);
assert_eq!(root.authority, "auth");
assert!(root.parent_id.is_none());
assert!(root.verify_integrity());
}
let _ = std::fs::remove_file(&db_path); }
#[test]
fn test_recovery_revision_history() {
use std::sync::Arc;
let db_path = std::env::temp_dir().join("test_recovery_history.log");
let _ = std::fs::remove_file(&db_path);
let rev1_id;
let rev2_id;
let rev3_id;
{
let db = crate::FeltDb::open(&db_path).expect("open db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let rev1 = store
.create("docs:1", r#"{"v":1}"#.to_string(), "auth".to_string())
.expect("create rev1");
rev1_id = rev1.id.clone();
let rev2 = store
.commit(r#"{"v":2}"#.to_string(), &rev1, "auth".to_string())
.expect("create rev2");
rev2_id = rev2.id.clone();
let rev3 = store
.commit(r#"{"v":3}"#.to_string(), &rev2, "auth".to_string())
.expect("create rev3");
rev3_id = rev3.id.clone();
assert_eq!(store.get(&rev3_id).unwrap().id, rev3_id);
assert_eq!(store.parent(&rev3_id).unwrap().id, rev2_id);
assert_eq!(store.parent(&rev2_id).unwrap().id, rev1_id);
}
{
let db = crate::FeltDb::open(&db_path).expect("open db again");
let store =
StateStore::with_feltdb(Arc::new(db)).expect("create store from recovered db");
assert!(store.exists(&rev1_id));
assert!(store.exists(&rev2_id));
assert!(store.exists(&rev3_id));
assert_eq!(store.get(&rev3_id).unwrap().id, rev3_id);
assert_eq!(store.parent(&rev3_id).unwrap().id, rev2_id);
assert_eq!(store.parent(&rev2_id).unwrap().id, rev1_id);
assert!(store.parent(&rev1_id).is_none());
assert_eq!(store.get(&rev1_id).unwrap().content, r#"{"v":1}"#);
assert_eq!(store.get(&rev2_id).unwrap().content, r#"{"v":2}"#);
assert_eq!(store.get(&rev3_id).unwrap().content, r#"{"v":3}"#);
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn test_divergent_siblings_survive_restart_independently() {
use std::sync::Arc;
let db_path = std::env::temp_dir().join("test_recovery_branches.log");
let _ = std::fs::remove_file(&db_path);
let root_id;
let branch1_id;
{
let db = crate::FeltDb::open(&db_path).expect("open db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let root = store
.create("docs:1", r#"{"base":true}"#.to_string(), "auth".to_string())
.expect("create root");
root_id = root.id.clone();
let branch1 = store
.commit(r#"{"path":"a"}"#.to_string(), &root, "auth".to_string())
.expect("create branch1");
branch1_id = branch1.id.clone();
}
{
let db = crate::FeltDb::open(&db_path).expect("open db again");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store over same log");
assert!(store.exists(&root_id));
assert!(store.exists(&branch1_id));
assert_eq!(store.parent(&branch1_id).unwrap().id, root_id);
assert_eq!(store.get(&branch1_id).unwrap().content, r#"{"path":"a"}"#);
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn test_every_revision_stays_retrievable_by_identity_after_restart() {
use std::sync::Arc;
let db_path = std::env::temp_dir().join("test_recovery_current.log");
let _ = std::fs::remove_file(&db_path);
let rev1_id;
let rev2_id;
{
let db = crate::FeltDb::open(&db_path).expect("open db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let rev1 = store
.create("docs:1", r#"{"state":"1"}"#.to_string(), "auth".to_string())
.expect("create rev1");
rev1_id = rev1.id.clone();
assert_eq!(store.get(&rev1_id).unwrap().id, rev1_id);
let rev2 = store
.commit(r#"{"state":"2"}"#.to_string(), &rev1, "auth".to_string())
.expect("create rev2");
rev2_id = rev2.id.clone();
assert_eq!(store.get(&rev2_id).unwrap().id, rev2_id);
assert_eq!(store.get(&rev1_id).unwrap().id, rev1_id);
}
{
let db = crate::FeltDb::open(&db_path).expect("open db again");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store over same log");
assert_eq!(store.get(&rev1_id).unwrap().content, r#"{"state":"1"}"#);
assert_eq!(store.get(&rev2_id).unwrap().content, r#"{"state":"2"}"#);
assert_eq!(store.parent(&rev2_id).unwrap().id, rev1_id);
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn test_recovery_complex_topology() {
use std::sync::Arc;
let db_path = std::env::temp_dir().join("test_recovery_complex.log");
let _ = std::fs::remove_file(&db_path);
let root_id;
let rev_a_id;
let rev_b_id;
{
let db = crate::FeltDb::open(&db_path).expect("open db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let root = store
.create("docs:1", r#"{"v":0}"#.to_string(), "auth".to_string())
.expect("create root");
root_id = root.id.clone();
let rev_a = store
.commit(
r#"{"v":1,"stage":"a"}"#.to_string(),
&root,
"auth".to_string(),
)
.expect("create rev_a");
rev_a_id = rev_a.id.clone();
let rev_b = store
.commit(
r#"{"v":2,"stage":"b"}"#.to_string(),
&rev_a,
"auth".to_string(),
)
.expect("create rev_b");
rev_b_id = rev_b.id.clone();
}
{
let db = crate::FeltDb::open(&db_path).expect("open db again");
let store =
StateStore::with_feltdb(Arc::new(db)).expect("create store from recovered db");
assert!(store.exists(&root_id));
assert!(store.exists(&rev_a_id));
assert!(store.exists(&rev_b_id));
assert_eq!(store.parent(&rev_b_id).unwrap().id, rev_a_id);
assert_eq!(store.parent(&rev_a_id).unwrap().id, root_id);
assert_eq!(
store.get(&rev_a_id).unwrap().content,
r#"{"v":1,"stage":"a"}"#
);
assert_eq!(
store.get(&rev_b_id).unwrap().content,
r#"{"v":2,"stage":"b"}"#
);
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn test_revisions_survive_multiple_restart_cycles() {
use std::sync::Arc;
let db_path = std::env::temp_dir().join("test_recovery_cycles.log");
let _ = std::fs::remove_file(&db_path);
let rev1_id;
let rev2_id;
{
let db = crate::FeltDb::open(&db_path).expect("open db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
rev1_id = store
.create("docs:1", r#"{"cycle":1}"#.to_string(), "auth".to_string())
.expect("create")
.id;
}
{
let db = crate::FeltDb::open(&db_path).expect("open db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let rev1 = store.get(&rev1_id).expect("rev1 survives");
assert_eq!(rev1.content, r#"{"cycle":1}"#);
rev2_id = store
.commit(r#"{"cycle":2}"#.to_string(), &rev1, "auth".to_string())
.expect("commit 2")
.id;
}
{
let db = crate::FeltDb::open(&db_path).expect("open db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
assert_eq!(store.get(&rev2_id).unwrap().content, r#"{"cycle":2}"#);
assert_eq!(store.parent(&rev2_id).unwrap().id, rev1_id);
}
{
let db = crate::FeltDb::open(&db_path).expect("open db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let rev2 = store.get(&rev2_id).expect("rev2 survives every cycle");
assert_eq!(rev2.content, r#"{"cycle":2}"#);
assert_eq!(rev2.parent_id, Some(rev1_id.clone()));
assert!(store.parent(&rev1_id).is_none(), "the root has no parent");
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn pr18_proof_single_write_survives_restart() {
let db_path = std::env::temp_dir().join("pr18_single_write.log");
let _ = std::fs::remove_file(&db_path);
let written_id = {
let db = crate::FeltDb::open(&db_path).expect("create db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let rev = store
.create(
"docs:1",
r#"{"task":"write PR18 proof"}"#.to_string(),
"system".to_string(),
)
.expect("write");
rev.id.clone()
};
{
let db = crate::FeltDb::open(&db_path).expect("reopen db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("recover store");
let recovered = store
.get(&written_id)
.expect("State should exist after restart");
assert_eq!(recovered.id, written_id, "State ID should match");
assert_eq!(
recovered.content, r#"{"task":"write PR18 proof"}"#,
"State content should survive restart"
);
assert!(
recovered.verify_integrity(),
"content must still hash to its identity"
);
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn pr18_proof_multi_record_atomicity() {
let db_path = std::env::temp_dir().join("pr18_multi_record.log");
let _ = std::fs::remove_file(&db_path);
let (project_id, membership_id, event_id) = {
let db = crate::FeltDb::open(&db_path).expect("create db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let project = store
.create(
"docs:1",
r#"{"type":"project","name":"FeltDB"}"#.to_string(),
"system".to_string(),
)
.expect("create project");
let membership = store
.commit(
r#"{"type":"membership","project":"FeltDB","user":"alice"}"#.to_string(),
&project,
"system".to_string(),
)
.expect("create membership");
let event = store
.commit(
r#"{"type":"audit","action":"project_created"}"#.to_string(),
&membership,
"system".to_string(),
)
.expect("create event");
(project.id.clone(), membership.id.clone(), event.id.clone())
};
{
let db = crate::FeltDb::open(&db_path).expect("reopen db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("recover store");
let event = store.get(&event_id).expect("event should exist");
assert_eq!(event.id, event_id);
assert_eq!(
event.parent_id,
Some(membership_id.clone()),
"Event parent should be membership"
);
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn pr18_proof_ancestry_integrity_on_restart() {
let db_path = std::env::temp_dir().join("pr18_ancestry.log");
let _ = std::fs::remove_file(&db_path);
let (root_id, child_id, grandchild_id) = {
let db = crate::FeltDb::open(&db_path).expect("create db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let root = store
.create(
"docs:1",
r#"{"generation":"root"}"#.to_string(),
"genesis".to_string(),
)
.expect("create root");
let child = store
.commit(
r#"{"generation":"child"}"#.to_string(),
&root,
"genesis".to_string(),
)
.expect("create child");
let grandchild = store
.commit(
r#"{"generation":"grandchild"}"#.to_string(),
&child,
"genesis".to_string(),
)
.expect("create grandchild");
(root.id.clone(), child.id.clone(), grandchild.id.clone())
};
{
let db = crate::FeltDb::open(&db_path).expect("reopen db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("recover store");
let grandchild = store.get(&grandchild_id).expect("grandchild exists");
assert_eq!(grandchild.id, grandchild_id);
let mut ancestry_count = 0;
let mut current_id = Some(grandchild.id.clone());
while let Some(id) = current_id {
ancestry_count += 1;
if id == root_id {
break; }
current_id = None; }
assert!(
ancestry_count > 0,
"Ancestry chain should be reconstructible"
);
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn pr18_proof_multiple_restart_cycles() {
let db_path = std::env::temp_dir().join("pr18_multi_cycle.log");
let _ = std::fs::remove_file(&db_path);
let mut state_id = {
let db = crate::FeltDb::open(&db_path).expect("create db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let rev = store
.create("docs:1", r#"{"cycle":1}"#.to_string(), "system".to_string())
.expect("create initial");
rev.id.clone()
};
for cycle in 2..=4 {
let db = crate::FeltDb::open(&db_path).expect("reopen db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("recover store");
let previous = store.get(&state_id).expect("previous state should exist");
assert_eq!(previous.id, state_id);
let rev = store
.commit(
format!(r#"{{"cycle":{}}}"#, cycle),
&previous,
"system".to_string(),
)
.expect("create update");
state_id = rev.id.clone();
}
{
let db = crate::FeltDb::open(&db_path).expect("final reopen");
let store = StateStore::with_feltdb(Arc::new(db)).expect("final recovery");
let final_state = store.get(&state_id).expect("final state exists");
assert_eq!(final_state.id, state_id, "Final state should match");
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn pr18_proof_no_partial_state_on_crash() {
let db_path = std::env::temp_dir().join("pr18_no_partial.log");
let _ = std::fs::remove_file(&db_path);
let initial_id = {
let db = crate::FeltDb::open(&db_path).expect("create db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let rev = store
.create(
"docs:1",
r#"{"status":"initial"}"#.to_string(),
"system".to_string(),
)
.expect("create initial");
rev.id.clone()
};
{
let db = crate::FeltDb::open(&db_path).expect("reopen");
let store = StateStore::with_feltdb(Arc::new(db)).expect("recover");
if let Some(recovered) = store.get(&initial_id) {
assert!(
recovered.content.contains("status"),
"State must be complete"
);
assert!(
recovered.content.contains("initial"),
"Attributes must be present"
);
assert_eq!(recovered.id, initial_id, "ID must match");
assert!(recovered.verify_integrity(), "and never partially written");
} else {
}
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn pr18_proof_identity_verification_on_recovery() {
let db_path = std::env::temp_dir().join("pr18_identity.log");
let _ = std::fs::remove_file(&db_path);
let expected_content = r#"{"integrity":"verified"}"#;
let state_id = {
let db = crate::FeltDb::open(&db_path).expect("create db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let rev = store
.create(
"docs:1",
expected_content.to_string(),
"integrity_check".to_string(),
)
.expect("create");
rev.id.clone()
};
{
let db = crate::FeltDb::open(&db_path).expect("reopen");
let store = StateStore::with_feltdb(Arc::new(db)).expect("recover");
let recovered = store.get(&state_id).expect("state exists");
assert_eq!(recovered.content, expected_content, "Content should match");
assert_eq!(recovered.id, state_id, "ID should match");
assert!(
recovered.verify_integrity(),
"Content hash should still be valid"
);
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn pr19_proof_concurrent_write_safety() {
use std::sync::{Arc, Mutex};
use std::thread;
let db_path = std::env::temp_dir().join("pr19_concurrent_writes.log");
let _ = std::fs::remove_file(&db_path);
let db = crate::FeltDb::open(&db_path).expect("create db");
let store = Arc::new(StateStore::with_feltdb(Arc::new(db)).expect("create store"));
let write_count = Arc::new(Mutex::new(0usize));
let mut handles = vec![];
for i in 0..5 {
let store_clone = Arc::clone(&store);
let count_clone = Arc::clone(&write_count);
let handle = thread::spawn(move || {
let content = format!(r#"{{"writer":{},"timestamp":{}}}"#, i, 1000 + i);
let rev = store_clone
.create(&format!("writer:{i}"), content, format!("writer_{}", i))
.expect("write should succeed");
let mut count = count_clone.lock().unwrap();
*count += 1;
rev.id.clone()
});
handles.push(handle);
}
let mut written_ids = vec![];
for handle in handles {
let id = handle.join().expect("thread should complete");
written_ids.push(id);
}
for id in &written_ids {
let written = store
.get(id)
.expect("every concurrently written revision should exist");
assert!(
written.verify_integrity(),
"no writer produced a partial revision"
);
}
let final_write = store
.create(
"docs:1",
r#"{"final":"write","success":true}"#.to_string(),
"system".to_string(),
)
.expect("final write should succeed");
assert!(
final_write.verify_integrity(),
"Final write should be valid"
);
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn pr19_proof_crash_during_sync_safety() {
let db_path = std::env::temp_dir().join("pr19_crash_during_sync.log");
let _ = std::fs::remove_file(&db_path);
let initial_id = {
let db = crate::FeltDb::open(&db_path).expect("create db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let rev = store
.create(
"docs:1",
r#"{"status":"pre-crash","integrity":"good"}"#.to_string(),
"system".to_string(),
)
.expect("write");
rev.id.clone()
};
{
let db = crate::FeltDb::open(&db_path).expect("reopen");
let store = StateStore::with_feltdb(Arc::new(db)).expect("recover");
let current = store
.get(&initial_id)
.expect("the initial state should still exist");
assert!(
current.verify_integrity(),
"State should be internally consistent"
);
assert!(
current.id == initial_id || current.verify_integrity(),
"Recovered state must be either initial (sync failed) or new+valid (sync succeeded)"
);
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn pr19_proof_query_index_selection() {
let db_path = std::env::temp_dir().join("pr19_query_index.log");
let _ = std::fs::remove_file(&db_path);
let db = crate::FeltDb::open(&db_path).expect("create db");
let store = Arc::new(StateStore::with_feltdb(Arc::new(db)).expect("create store"));
for i in 0..100 {
let content = format!(r#"{{"user_id":"user_{}","active":true}}"#, i % 10);
let _ = store
.create(&format!("record:{i}"), content, format!("creator_{}", i))
.expect("write record");
}
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn pr19_proof_multi_tenant_isolation_basic() {
let db_path = std::env::temp_dir().join("pr19_multi_tenant.log");
let _ = std::fs::remove_file(&db_path);
let db = crate::FeltDb::open(&db_path).expect("create db");
let store = Arc::new(StateStore::with_feltdb(Arc::new(db)).expect("create store"));
let tenant_a_record = store
.create(
"tenant_a:1",
r#"{"tenant":"tenant_a","data":"secret_a"}"#.to_string(),
"tenant_a".to_string(),
)
.expect("tenant_a write");
let tenant_b_record = store
.create(
"tenant_b:1",
r#"{"tenant":"tenant_b","data":"secret_b"}"#.to_string(),
"tenant_b".to_string(),
)
.expect("tenant_b write");
assert_ne!(
tenant_a_record.id, tenant_b_record.id,
"Records should have different IDs"
);
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn pr19_proof_atomicity_requires_sync_return() {
let db_path = std::env::temp_dir().join("pr19_atomicity_sync.log");
let _ = std::fs::remove_file(&db_path);
let rev_id = {
let db = crate::FeltDb::open(&db_path).expect("create db");
let store = StateStore::with_feltdb(Arc::new(db)).expect("create store");
let result = store.create(
"docs:1",
r#"{"atomicity":"requires_sync_success"}"#.to_string(),
"system".to_string(),
);
match result {
Ok(rev) => Some(rev.id.clone()),
Err(_) => None,
}
};
if let Some(rev_id) = rev_id {
let db2 = crate::FeltDb::open(&db_path).expect("reopen");
let store2 = StateStore::with_feltdb(Arc::new(db2)).expect("recover");
assert_eq!(
store2.get(&rev_id).unwrap().id,
rev_id,
"Durability guaranteed"
);
}
let _ = std::fs::remove_file(&db_path);
}
}