feat: schema projections, command transactions, quantum-secure encryption
This commit is contained in:
Will Anderson
@@ -0,0 +1,17 @@
|
||||
[package]
|
||||
name = "engram-tx"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
description = "Command pattern + rollback-of-rollback transaction engine for Engram"
|
||||
license = "MIT"
|
||||
|
||||
[dependencies]
|
||||
engram-core = { path = "../engram-core" }
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_json = "1"
|
||||
uuid = { version = "1", features = ["v4", "serde"] }
|
||||
thiserror = "1"
|
||||
sled = "0.34"
|
||||
|
||||
[dev-dependencies]
|
||||
tempfile = "3"
|
||||
@@ -0,0 +1,166 @@
|
||||
/// Command — the first-class mutation unit of Engram's transaction system.
|
||||
///
|
||||
/// A command is an immutable record of intent. Once created, its ID, type,
|
||||
/// idempotency key, and causal parent never change. Status and conflict fields
|
||||
/// are the only mutable parts, updated as the command moves through its lifecycle.
|
||||
use engram_core::types::MemoryTier;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::Value;
|
||||
use uuid::Uuid;
|
||||
|
||||
/// The operation a command performs.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub enum CommandType {
|
||||
/// Create a new node. `payload` contains the node's fields.
|
||||
CreateNode,
|
||||
/// Update an existing node's content/metadata. `payload` contains the new fields.
|
||||
UpdateNode,
|
||||
/// Delete a node by UUID.
|
||||
DeleteNode,
|
||||
/// Create an edge between two nodes.
|
||||
CreateEdge,
|
||||
/// Delete an edge by its from/to pair.
|
||||
DeleteEdge,
|
||||
/// Update a node's salience score.
|
||||
UpdateSalience,
|
||||
/// Batch import of many nodes/edges.
|
||||
BulkImport,
|
||||
/// Roll back a previously applied command. The UUID is the target command's ID.
|
||||
Rollback(Uuid),
|
||||
}
|
||||
|
||||
/// Lifecycle status of a command.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub enum CommandStatus {
|
||||
/// Created but not yet applied.
|
||||
Pending,
|
||||
/// Successfully applied to the database.
|
||||
Applied,
|
||||
/// Rolled back (a subsequent Rollback command was applied).
|
||||
RolledBack,
|
||||
/// Applied but conflicted with another command; conflict details in the log.
|
||||
Conflicted,
|
||||
}
|
||||
|
||||
/// A command in Engram's append-only mutation log.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Command {
|
||||
/// Stable unique identifier for this command.
|
||||
pub id: Uuid,
|
||||
/// The operation this command performs.
|
||||
pub command_type: CommandType,
|
||||
/// The forward operation data — what to do.
|
||||
pub payload: Value,
|
||||
/// The inverse operation data — computed eagerly at command creation time.
|
||||
/// This is the "undo" data, available even if the graph has changed since.
|
||||
pub inverse_payload: Value,
|
||||
/// Dedup key. If a command with this idempotency key has already been applied,
|
||||
/// the new command is a no-op. Prevents double-application in distributed sync.
|
||||
pub idempotency_key: String,
|
||||
/// The command that caused this one, if any (forms the causal DAG).
|
||||
pub causal_parent: Option<Uuid>,
|
||||
/// Unix milliseconds when this command was created.
|
||||
pub timestamp_ms: i64,
|
||||
/// Current lifecycle status.
|
||||
pub status: CommandStatus,
|
||||
/// Which peer originated this command (for conflict resolution).
|
||||
pub peer_id: Option<Uuid>,
|
||||
/// If conflicted, a human-readable description of the conflict.
|
||||
pub conflict_note: Option<String>,
|
||||
}
|
||||
|
||||
impl Command {
|
||||
/// Create a new pending command.
|
||||
pub fn new(
|
||||
command_type: CommandType,
|
||||
payload: Value,
|
||||
inverse_payload: Value,
|
||||
idempotency_key: impl Into<String>,
|
||||
causal_parent: Option<Uuid>,
|
||||
peer_id: Option<Uuid>,
|
||||
) -> Self {
|
||||
Self {
|
||||
id: Uuid::new_v4(),
|
||||
command_type,
|
||||
payload,
|
||||
inverse_payload,
|
||||
idempotency_key: idempotency_key.into(),
|
||||
causal_parent,
|
||||
timestamp_ms: engram_core::types::now_ms(),
|
||||
status: CommandStatus::Pending,
|
||||
peer_id,
|
||||
conflict_note: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The result of successfully applying a command.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct CommandResult {
|
||||
/// The command that was applied.
|
||||
pub command_id: Uuid,
|
||||
/// The command's new status.
|
||||
pub status: CommandStatus,
|
||||
/// Any entity UUID produced by the operation (e.g., the new node's UUID).
|
||||
pub produced_id: Option<Uuid>,
|
||||
/// Whether this was a no-op due to idempotency.
|
||||
pub was_idempotent: bool,
|
||||
}
|
||||
|
||||
// ── Payload schema helpers ────────────────────────────────────────────────────
|
||||
|
||||
/// Payload for CreateNode commands.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct CreateNodePayload {
|
||||
pub node_id: Uuid,
|
||||
pub node_type: String,
|
||||
pub embedding: Vec<f32>,
|
||||
pub content: Vec<u8>,
|
||||
pub tier: MemoryTier,
|
||||
pub importance: f32,
|
||||
}
|
||||
|
||||
/// Payload for UpdateNode commands.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct UpdateNodePayload {
|
||||
pub node_id: Uuid,
|
||||
pub new_content: Option<Vec<u8>>,
|
||||
pub new_importance: Option<f32>,
|
||||
pub new_tier: Option<MemoryTier>,
|
||||
}
|
||||
|
||||
/// Payload for DeleteNode commands.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct DeleteNodePayload {
|
||||
pub node_id: Uuid,
|
||||
/// The full node is saved at command-creation time so rollback can restore it.
|
||||
pub snapshot: Value,
|
||||
}
|
||||
|
||||
/// Payload for CreateEdge commands.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct CreateEdgePayload {
|
||||
pub edge_id: Uuid,
|
||||
pub from_id: Uuid,
|
||||
pub to_id: Uuid,
|
||||
pub relation: String,
|
||||
pub weight: f32,
|
||||
}
|
||||
|
||||
/// Payload for DeleteEdge commands.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct DeleteEdgePayload {
|
||||
pub edge_id: Uuid,
|
||||
pub from_id: Uuid,
|
||||
pub to_id: Uuid,
|
||||
/// Full edge snapshot for rollback.
|
||||
pub snapshot: Value,
|
||||
}
|
||||
|
||||
/// Payload for UpdateSalience commands.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct UpdateSaliencePayload {
|
||||
pub node_id: Uuid,
|
||||
pub new_salience: f32,
|
||||
pub old_salience: f32,
|
||||
}
|
||||
@@ -0,0 +1,710 @@
|
||||
/// TransactionEngine — applies commands to EngramDb and manages rollback.
|
||||
///
|
||||
/// The engine wraps an `EngramDb` and a `CommandLog`. All mutations go through
|
||||
/// `apply()`. Rollbacks create new inverse commands. Rolling back a rollback
|
||||
/// re-applies the original — full undo/redo with causal tracking.
|
||||
use engram_core::types::{Edge, Node, NodeType, RelationType};
|
||||
use engram_core::EngramDb;
|
||||
use serde_json::Value;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::command::{
|
||||
Command, CommandResult, CommandStatus, CommandType, CreateEdgePayload, CreateNodePayload,
|
||||
DeleteEdgePayload, DeleteNodePayload, UpdateNodePayload, UpdateSaliencePayload,
|
||||
};
|
||||
use crate::error::{TxError, TxResult};
|
||||
use crate::log::CommandLog;
|
||||
|
||||
pub struct TransactionEngine {
|
||||
db: std::sync::Arc<std::sync::Mutex<EngramDb>>,
|
||||
log: CommandLog,
|
||||
/// Our peer ID for conflict resolution.
|
||||
peer_id: Option<Uuid>,
|
||||
}
|
||||
|
||||
impl TransactionEngine {
|
||||
/// Create a new engine backed by the given database and a sled store for the log.
|
||||
pub fn new(
|
||||
db: std::sync::Arc<std::sync::Mutex<EngramDb>>,
|
||||
log_db: sled::Db,
|
||||
peer_id: Option<Uuid>,
|
||||
) -> Self {
|
||||
Self {
|
||||
db,
|
||||
log: CommandLog::open(log_db),
|
||||
peer_id,
|
||||
}
|
||||
}
|
||||
|
||||
// ── Public API ────────────────────────────────────────────────────────────
|
||||
|
||||
/// Apply a command to the database.
|
||||
///
|
||||
/// Idempotency: if a command with the same `idempotency_key` has already
|
||||
/// been applied, this is a no-op and returns the original command's result.
|
||||
pub fn apply(&mut self, mut cmd: Command) -> TxResult<CommandResult> {
|
||||
// Idempotency check
|
||||
if let Some(existing_id) = self.log.check_idempotency(&cmd.idempotency_key)? {
|
||||
return Ok(CommandResult {
|
||||
command_id: existing_id,
|
||||
status: CommandStatus::Applied,
|
||||
produced_id: None,
|
||||
was_idempotent: true,
|
||||
});
|
||||
}
|
||||
|
||||
// Execute the operation
|
||||
let produced_id = self.execute(&cmd)?;
|
||||
|
||||
cmd.status = CommandStatus::Applied;
|
||||
self.log.write(&cmd)?;
|
||||
|
||||
Ok(CommandResult {
|
||||
command_id: cmd.id,
|
||||
status: CommandStatus::Applied,
|
||||
produced_id,
|
||||
was_idempotent: false,
|
||||
})
|
||||
}
|
||||
|
||||
/// Roll back a previously applied command.
|
||||
///
|
||||
/// Creates and applies a new `Rollback(target_id)` command whose payload
|
||||
/// is the inverse of the target command. The target command's status is
|
||||
/// updated to `RolledBack`.
|
||||
///
|
||||
/// Returns the new rollback command (useful for tracking / further rollback).
|
||||
pub fn rollback(&mut self, target_id: Uuid) -> TxResult<Command> {
|
||||
let target = self.log.require(target_id)?;
|
||||
|
||||
if target.status == CommandStatus::RolledBack {
|
||||
return Err(TxError::InvalidStatus(format!(
|
||||
"command {} is already rolled back",
|
||||
target_id
|
||||
)));
|
||||
}
|
||||
if target.status == CommandStatus::Pending {
|
||||
return Err(TxError::InvalidStatus(
|
||||
"cannot roll back a pending command".into(),
|
||||
));
|
||||
}
|
||||
|
||||
// The rollback's payload is the original command's inverse_payload.
|
||||
// The rollback's inverse_payload is the original command's payload.
|
||||
// This enables rollback-of-rollback to re-apply the original.
|
||||
let rollback_key = format!("rollback:{}", target_id);
|
||||
let rollback_cmd = Command::new(
|
||||
CommandType::Rollback(target_id),
|
||||
target.inverse_payload.clone(),
|
||||
target.payload.clone(),
|
||||
rollback_key,
|
||||
Some(target_id),
|
||||
self.peer_id,
|
||||
);
|
||||
|
||||
// Execute the inverse operation
|
||||
self.execute_inverse(&target)?;
|
||||
|
||||
// Mark the original command as rolled back
|
||||
let mut updated_target = target;
|
||||
updated_target.status = CommandStatus::RolledBack;
|
||||
self.log.write(&updated_target)?;
|
||||
|
||||
// Persist the rollback command as Applied
|
||||
let mut rb = rollback_cmd;
|
||||
rb.status = CommandStatus::Applied;
|
||||
self.log.write(&rb)?;
|
||||
|
||||
Ok(rb)
|
||||
}
|
||||
|
||||
/// Roll back a rollback — re-applying the original command.
|
||||
///
|
||||
/// This is "undo the undo". The rollback_id must be a command of type
|
||||
/// `Rollback(original_id)`. Rolling it back re-applies `original_id`.
|
||||
pub fn rollback_rollback(&mut self, rollback_id: Uuid) -> TxResult<Command> {
|
||||
let rb_cmd = self.log.require(rollback_id)?;
|
||||
|
||||
// Verify this IS a rollback command
|
||||
let original_id = match &rb_cmd.command_type {
|
||||
CommandType::Rollback(orig) => *orig,
|
||||
_ => {
|
||||
return Err(TxError::Invalid(format!(
|
||||
"command {} is not a Rollback command",
|
||||
rollback_id
|
||||
)));
|
||||
}
|
||||
};
|
||||
|
||||
if rb_cmd.status == CommandStatus::RolledBack {
|
||||
return Err(TxError::InvalidStatus(
|
||||
"this rollback has itself already been rolled back".into(),
|
||||
));
|
||||
}
|
||||
|
||||
// The rollback_rollback's payload is rb_cmd.inverse_payload (= original payload)
|
||||
// Its inverse is rb_cmd.payload (= original's inverse_payload)
|
||||
let key = format!("rollback:{}", rollback_id);
|
||||
let rr_cmd = Command::new(
|
||||
CommandType::Rollback(rollback_id),
|
||||
rb_cmd.inverse_payload.clone(),
|
||||
rb_cmd.payload.clone(),
|
||||
key,
|
||||
Some(rollback_id),
|
||||
self.peer_id,
|
||||
);
|
||||
|
||||
// Re-apply the original command by executing the original's payload
|
||||
let original = self.log.require(original_id)?;
|
||||
self.execute(&original)?;
|
||||
|
||||
// Mark original as Applied again
|
||||
let mut orig = original;
|
||||
orig.status = CommandStatus::Applied;
|
||||
self.log.write(&orig)?;
|
||||
|
||||
// Mark rollback as RolledBack
|
||||
let mut rb = rb_cmd;
|
||||
rb.status = CommandStatus::RolledBack;
|
||||
self.log.write(&rb)?;
|
||||
|
||||
// Persist the new re-apply command
|
||||
let mut rr = rr_cmd;
|
||||
rr.status = CommandStatus::Applied;
|
||||
self.log.write(&rr)?;
|
||||
|
||||
Ok(rr)
|
||||
}
|
||||
|
||||
/// All commands since a given Unix millisecond timestamp.
|
||||
pub fn history(&self, since_ms: i64) -> TxResult<Vec<Command>> {
|
||||
self.log.since(since_ms)
|
||||
}
|
||||
|
||||
/// The causal chain for a given command (root-first).
|
||||
pub fn causal_chain(&self, command_id: Uuid) -> TxResult<Vec<Command>> {
|
||||
self.log.causal_chain(command_id)
|
||||
}
|
||||
|
||||
/// Access the command log directly (for server routes).
|
||||
pub fn log(&self) -> &CommandLog {
|
||||
&self.log
|
||||
}
|
||||
|
||||
// ── Command execution ─────────────────────────────────────────────────────
|
||||
|
||||
fn execute(&self, cmd: &Command) -> TxResult<Option<Uuid>> {
|
||||
match &cmd.command_type {
|
||||
CommandType::CreateNode => self.exec_create_node(&cmd.payload),
|
||||
CommandType::UpdateNode => {
|
||||
self.exec_update_node(&cmd.payload)?;
|
||||
Ok(None)
|
||||
}
|
||||
CommandType::DeleteNode => {
|
||||
self.exec_delete_node(&cmd.payload)?;
|
||||
Ok(None)
|
||||
}
|
||||
CommandType::CreateEdge => {
|
||||
self.exec_create_edge(&cmd.payload)?;
|
||||
Ok(None)
|
||||
}
|
||||
CommandType::DeleteEdge => {
|
||||
self.exec_delete_edge(&cmd.payload)?;
|
||||
Ok(None)
|
||||
}
|
||||
CommandType::UpdateSalience => {
|
||||
self.exec_update_salience(&cmd.payload)?;
|
||||
Ok(None)
|
||||
}
|
||||
CommandType::BulkImport => {
|
||||
self.exec_bulk_import(&cmd.payload)?;
|
||||
Ok(None)
|
||||
}
|
||||
CommandType::Rollback(_) => {
|
||||
// Rollback commands are executed via execute_inverse on the target
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn execute_inverse(&self, cmd: &Command) -> TxResult<()> {
|
||||
// The inverse is described by inverse_payload, which mirrors the command type
|
||||
match &cmd.command_type {
|
||||
CommandType::CreateNode => {
|
||||
// Inverse of CreateNode is DeleteNode
|
||||
let p: CreateNodePayload = serde_json::from_value(cmd.payload.clone())?;
|
||||
let db = self.db.lock().map_err(|_| TxError::Invalid("lock poisoned".into()))?;
|
||||
db.delete_node(p.node_id)?;
|
||||
}
|
||||
CommandType::DeleteNode => {
|
||||
// Inverse of DeleteNode is re-creating the node from snapshot
|
||||
let p: DeleteNodePayload = serde_json::from_value(cmd.payload.clone())?;
|
||||
let node: Node = serde_json::from_value(p.snapshot)?;
|
||||
let db = self.db.lock().map_err(|_| TxError::Invalid("lock poisoned".into()))?;
|
||||
db.put_node(node)?;
|
||||
}
|
||||
CommandType::UpdateNode => {
|
||||
// Inverse is applying the inverse_payload (prior state)
|
||||
self.exec_update_node(&cmd.inverse_payload)?;
|
||||
}
|
||||
CommandType::CreateEdge => {
|
||||
// Inverse of CreateEdge is DeleteEdge
|
||||
let p: CreateEdgePayload = serde_json::from_value(cmd.payload.clone())?;
|
||||
self.delete_edge_by_pair(p.from_id, p.to_id)?;
|
||||
}
|
||||
CommandType::DeleteEdge => {
|
||||
// Inverse of DeleteEdge is re-creating the edge from snapshot
|
||||
let p: DeleteEdgePayload = serde_json::from_value(cmd.payload.clone())?;
|
||||
let edge: Edge = serde_json::from_value(p.snapshot)?;
|
||||
let db = self.db.lock().map_err(|_| TxError::Invalid("lock poisoned".into()))?;
|
||||
db.put_edge(edge)?;
|
||||
}
|
||||
CommandType::UpdateSalience => {
|
||||
// Restore old salience
|
||||
let p: UpdateSaliencePayload = serde_json::from_value(cmd.payload.clone())?;
|
||||
let restore = UpdateSaliencePayload {
|
||||
node_id: p.node_id,
|
||||
new_salience: p.old_salience,
|
||||
old_salience: p.new_salience,
|
||||
};
|
||||
self.exec_update_salience(&serde_json::to_value(restore)?)?;
|
||||
}
|
||||
CommandType::BulkImport | CommandType::Rollback(_) => {
|
||||
// BulkImport rollback would need to individually undo each item.
|
||||
// For now, we store the inverse_payload as instructions and log the gap.
|
||||
// TODO: implement fine-grained BulkImport rollback
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ── Operation implementations ─────────────────────────────────────────────
|
||||
|
||||
fn exec_create_node(&self, payload: &Value) -> TxResult<Option<Uuid>> {
|
||||
let p: CreateNodePayload = serde_json::from_value(payload.clone())?;
|
||||
let node_type = parse_node_type(&p.node_type)?;
|
||||
let tier = p.tier;
|
||||
let node = Node::new(node_type, p.embedding, p.content, tier, p.importance)
|
||||
.with_id(p.node_id);
|
||||
let db = self.db.lock().map_err(|_| TxError::Invalid("lock poisoned".into()))?;
|
||||
db.put_node(node)?;
|
||||
Ok(Some(p.node_id))
|
||||
}
|
||||
|
||||
fn exec_update_node(&self, payload: &Value) -> TxResult<()> {
|
||||
let p: UpdateNodePayload = serde_json::from_value(payload.clone())?;
|
||||
let db = self.db.lock().map_err(|_| TxError::Invalid("lock poisoned".into()))?;
|
||||
let mut node = db
|
||||
.get_node(p.node_id)?
|
||||
.ok_or(TxError::NotFound(p.node_id))?;
|
||||
if let Some(content) = p.new_content {
|
||||
node.content = content;
|
||||
}
|
||||
if let Some(importance) = p.new_importance {
|
||||
node.importance = importance.clamp(0.0, 1.0);
|
||||
}
|
||||
if let Some(tier) = p.new_tier {
|
||||
node.tier = tier;
|
||||
}
|
||||
db.put_node(node)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn exec_delete_node(&self, payload: &Value) -> TxResult<()> {
|
||||
let p: DeleteNodePayload = serde_json::from_value(payload.clone())?;
|
||||
let db = self.db.lock().map_err(|_| TxError::Invalid("lock poisoned".into()))?;
|
||||
db.delete_node(p.node_id)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn exec_create_edge(&self, payload: &Value) -> TxResult<()> {
|
||||
let p: CreateEdgePayload = serde_json::from_value(payload.clone())?;
|
||||
let relation = parse_relation(&p.relation)?;
|
||||
let edge = Edge::new(p.from_id, p.to_id, relation, p.weight);
|
||||
let db = self.db.lock().map_err(|_| TxError::Invalid("lock poisoned".into()))?;
|
||||
db.put_edge(edge)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn exec_delete_edge(&self, payload: &Value) -> TxResult<()> {
|
||||
let p: DeleteEdgePayload = serde_json::from_value(payload.clone())?;
|
||||
self.delete_edge_by_pair(p.from_id, p.to_id)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn exec_update_salience(&self, payload: &Value) -> TxResult<()> {
|
||||
let p: UpdateSaliencePayload = serde_json::from_value(payload.clone())?;
|
||||
let db = self.db.lock().map_err(|_| TxError::Invalid("lock poisoned".into()))?;
|
||||
let mut node = db
|
||||
.get_node(p.node_id)?
|
||||
.ok_or(TxError::NotFound(p.node_id))?;
|
||||
node.salience = p.new_salience;
|
||||
db.put_node(node)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn exec_bulk_import(&self, payload: &Value) -> TxResult<()> {
|
||||
let nodes_val = payload.get("nodes").and_then(|v| v.as_array()).cloned().unwrap_or_default();
|
||||
let edges_val = payload.get("edges").and_then(|v| v.as_array()).cloned().unwrap_or_default();
|
||||
|
||||
let db = self.db.lock().map_err(|_| TxError::Invalid("lock poisoned".into()))?;
|
||||
for nv in nodes_val {
|
||||
let node: Node = serde_json::from_value(nv)?;
|
||||
db.put_node(node)?;
|
||||
}
|
||||
for ev in edges_val {
|
||||
let edge: Edge = serde_json::from_value(ev)?;
|
||||
db.put_edge(edge)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn delete_edge_by_pair(&self, from_id: Uuid, to_id: Uuid) -> TxResult<()> {
|
||||
// sled stores edges at "edges:from:{from}:{to}" — we remove both directions
|
||||
let db = self.db.lock().map_err(|_| TxError::Invalid("lock poisoned".into()))?;
|
||||
// We can't directly call into storage here without re-exposing internals,
|
||||
// so we use the public scan-and-check approach via get_edges_from
|
||||
let edges = db.get_edges_from(from_id)?;
|
||||
for edge in edges {
|
||||
if edge.to_id == to_id {
|
||||
// Re-insert a tombstone isn't directly supported — we use the
|
||||
// internal sled key to delete. Since we don't have direct sled
|
||||
// access through EngramDb's public API, we rely on the fact that
|
||||
// put_edge with weight=0 effectively nullifies it, but for proper
|
||||
// deletion we need access to the underlying store.
|
||||
//
|
||||
// We work around this by storing a zero-weight edge with weight=-1
|
||||
// as a sentinel, or by exposing delete_edge. Since delete_node is
|
||||
// already exposed, we add a helper. For now, we store the edge with
|
||||
// weight 0 (marking it inactive) until a delete_edge API is added.
|
||||
//
|
||||
// TODO: add db.delete_edge() to engram-core public API.
|
||||
// For now: overwrite with weight 0 to effectively disable it.
|
||||
let mut tombstone = edge;
|
||||
tombstone.weight = 0.0;
|
||||
db.put_edge(tombstone)?;
|
||||
break;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
// ── Command builder helpers ───────────────────────────────────────────────────
|
||||
|
||||
/// Build a CreateNode command with eagerly-computed inverse.
|
||||
pub fn build_create_node_cmd(
|
||||
node: &Node,
|
||||
idempotency_key: impl Into<String>,
|
||||
peer_id: Option<Uuid>,
|
||||
) -> Command {
|
||||
let payload = serde_json::to_value(CreateNodePayload {
|
||||
node_id: node.id,
|
||||
node_type: node_type_to_str(&node.node_type).to_string(),
|
||||
embedding: node.embedding.clone(),
|
||||
content: node.content.clone(),
|
||||
tier: node.tier.clone(),
|
||||
importance: node.importance,
|
||||
})
|
||||
.unwrap_or(Value::Null);
|
||||
|
||||
// Inverse: delete the node we're about to create
|
||||
let inverse = serde_json::to_value(DeleteNodePayload {
|
||||
node_id: node.id,
|
||||
snapshot: serde_json::to_value(node).unwrap_or(Value::Null),
|
||||
})
|
||||
.unwrap_or(Value::Null);
|
||||
|
||||
Command::new(
|
||||
CommandType::CreateNode,
|
||||
payload,
|
||||
inverse,
|
||||
idempotency_key,
|
||||
None,
|
||||
peer_id,
|
||||
)
|
||||
}
|
||||
|
||||
/// Build a DeleteNode command with eagerly-computed inverse (snapshot).
|
||||
pub fn build_delete_node_cmd(
|
||||
node: &Node,
|
||||
idempotency_key: impl Into<String>,
|
||||
peer_id: Option<Uuid>,
|
||||
) -> Command {
|
||||
let snapshot = serde_json::to_value(node).unwrap_or(Value::Null);
|
||||
let payload = serde_json::to_value(DeleteNodePayload {
|
||||
node_id: node.id,
|
||||
snapshot: snapshot.clone(),
|
||||
})
|
||||
.unwrap_or(Value::Null);
|
||||
|
||||
// Inverse: re-create the node from snapshot
|
||||
let inverse = serde_json::to_value(CreateNodePayload {
|
||||
node_id: node.id,
|
||||
node_type: node_type_to_str(&node.node_type).to_string(),
|
||||
embedding: node.embedding.clone(),
|
||||
content: node.content.clone(),
|
||||
tier: node.tier.clone(),
|
||||
importance: node.importance,
|
||||
})
|
||||
.unwrap_or(Value::Null);
|
||||
|
||||
Command::new(
|
||||
CommandType::DeleteNode,
|
||||
payload,
|
||||
inverse,
|
||||
idempotency_key,
|
||||
None,
|
||||
peer_id,
|
||||
)
|
||||
}
|
||||
|
||||
/// Build a CreateEdge command.
|
||||
pub fn build_create_edge_cmd(
|
||||
edge: &Edge,
|
||||
idempotency_key: impl Into<String>,
|
||||
peer_id: Option<Uuid>,
|
||||
) -> Command {
|
||||
let relation_str = relation_to_str(&edge.relation).to_string();
|
||||
let payload = serde_json::to_value(CreateEdgePayload {
|
||||
edge_id: edge.id,
|
||||
from_id: edge.from_id,
|
||||
to_id: edge.to_id,
|
||||
relation: relation_str.clone(),
|
||||
weight: edge.weight,
|
||||
})
|
||||
.unwrap_or(Value::Null);
|
||||
|
||||
// Inverse: delete the edge
|
||||
let snapshot = serde_json::to_value(edge).unwrap_or(Value::Null);
|
||||
let inverse = serde_json::to_value(DeleteEdgePayload {
|
||||
edge_id: edge.id,
|
||||
from_id: edge.from_id,
|
||||
to_id: edge.to_id,
|
||||
snapshot,
|
||||
})
|
||||
.unwrap_or(Value::Null);
|
||||
|
||||
Command::new(
|
||||
CommandType::CreateEdge,
|
||||
payload,
|
||||
inverse,
|
||||
idempotency_key,
|
||||
None,
|
||||
peer_id,
|
||||
)
|
||||
}
|
||||
|
||||
// ── Type string helpers ───────────────────────────────────────────────────────
|
||||
|
||||
fn parse_node_type(s: &str) -> TxResult<NodeType> {
|
||||
match s {
|
||||
"Memory" => Ok(NodeType::Memory),
|
||||
"Concept" => Ok(NodeType::Concept),
|
||||
"Event" => Ok(NodeType::Event),
|
||||
"Entity" => Ok(NodeType::Entity),
|
||||
"Process" => Ok(NodeType::Process),
|
||||
"InternalState" => Ok(NodeType::InternalState),
|
||||
_ => Err(TxError::Invalid(format!("unknown node type: {}", s))),
|
||||
}
|
||||
}
|
||||
|
||||
fn node_type_to_str(t: &NodeType) -> &'static str {
|
||||
match t {
|
||||
NodeType::Memory => "Memory",
|
||||
NodeType::Concept => "Concept",
|
||||
NodeType::Event => "Event",
|
||||
NodeType::Entity => "Entity",
|
||||
NodeType::Process => "Process",
|
||||
NodeType::InternalState => "InternalState",
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_relation(s: &str) -> TxResult<RelationType> {
|
||||
match s {
|
||||
"Supersedes" => Ok(RelationType::Supersedes),
|
||||
"Causes" => Ok(RelationType::Causes),
|
||||
"Contains" => Ok(RelationType::Contains),
|
||||
"References" => Ok(RelationType::References),
|
||||
"Contradicts" => Ok(RelationType::Contradicts),
|
||||
"Exemplifies" => Ok(RelationType::Exemplifies),
|
||||
"Activates" => Ok(RelationType::Activates),
|
||||
"TemporallyPrecedes" => Ok(RelationType::TemporallyPrecedes),
|
||||
_ => Err(TxError::Invalid(format!("unknown relation: {}", s))),
|
||||
}
|
||||
}
|
||||
|
||||
fn relation_to_str(r: &RelationType) -> &'static str {
|
||||
match r {
|
||||
RelationType::Supersedes => "Supersedes",
|
||||
RelationType::Causes => "Causes",
|
||||
RelationType::Contains => "Contains",
|
||||
RelationType::References => "References",
|
||||
RelationType::Contradicts => "Contradicts",
|
||||
RelationType::Exemplifies => "Exemplifies",
|
||||
RelationType::Activates => "Activates",
|
||||
RelationType::TemporallyPrecedes => "TemporallyPrecedes",
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use engram_core::types::{MemoryTier, Node, NodeType};
|
||||
use tempfile::TempDir;
|
||||
|
||||
fn make_engine() -> (TransactionEngine, TempDir, TempDir) {
|
||||
let db_dir = TempDir::new().unwrap();
|
||||
let log_dir = TempDir::new().unwrap();
|
||||
let db = engram_core::EngramDb::open(db_dir.path()).unwrap();
|
||||
let db = std::sync::Arc::new(std::sync::Mutex::new(db));
|
||||
let log_db = sled::open(log_dir.path()).unwrap();
|
||||
let engine = TransactionEngine::new(db, log_db, None);
|
||||
(engine, db_dir, log_dir)
|
||||
}
|
||||
|
||||
fn make_node() -> Node {
|
||||
Node::new(
|
||||
NodeType::Memory,
|
||||
vec![1.0, 0.0],
|
||||
b"test content".to_vec(),
|
||||
MemoryTier::Semantic,
|
||||
0.8,
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_apply_create_node() {
|
||||
let (mut engine, _db_dir, _log_dir) = make_engine();
|
||||
let node = make_node();
|
||||
let cmd = build_create_node_cmd(&node, "create-test-1", None);
|
||||
let result = engine.apply(cmd).unwrap();
|
||||
assert_eq!(result.status, CommandStatus::Applied);
|
||||
assert!(!result.was_idempotent);
|
||||
|
||||
// Node should now exist
|
||||
let db = engine.db.lock().unwrap();
|
||||
let found = db.get_node(node.id).unwrap();
|
||||
assert!(found.is_some());
|
||||
assert_eq!(found.unwrap().content, b"test content");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_idempotency() {
|
||||
let (mut engine, _db_dir, _log_dir) = make_engine();
|
||||
let node = make_node();
|
||||
let cmd1 = build_create_node_cmd(&node, "idem-key-42", None);
|
||||
let cmd2 = build_create_node_cmd(&node, "idem-key-42", None);
|
||||
|
||||
let r1 = engine.apply(cmd1).unwrap();
|
||||
let r2 = engine.apply(cmd2).unwrap();
|
||||
|
||||
assert!(!r1.was_idempotent);
|
||||
assert!(r2.was_idempotent);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rollback_delete_node() {
|
||||
let (mut engine, _db_dir, _log_dir) = make_engine();
|
||||
let node = make_node();
|
||||
let node_id = node.id;
|
||||
|
||||
// First apply create
|
||||
let create_cmd = build_create_node_cmd(&node, "create-rb-1", None);
|
||||
let result = engine.apply(create_cmd).unwrap();
|
||||
|
||||
// Roll back the creation — should delete the node
|
||||
let rb = engine.rollback(result.command_id).unwrap();
|
||||
assert_eq!(rb.status, CommandStatus::Applied);
|
||||
|
||||
let db = engine.db.lock().unwrap();
|
||||
let found = db.get_node(node_id).unwrap();
|
||||
assert!(found.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_rollback_of_rollback() {
|
||||
let (mut engine, _db_dir, _log_dir) = make_engine();
|
||||
let node = make_node();
|
||||
let node_id = node.id;
|
||||
|
||||
// Create the node
|
||||
let create_cmd = build_create_node_cmd(&node, "create-rorb-1", None);
|
||||
let create_result = engine.apply(create_cmd).unwrap();
|
||||
|
||||
// Roll back (delete)
|
||||
let rb = engine.rollback(create_result.command_id).unwrap();
|
||||
|
||||
// Verify it's gone
|
||||
{
|
||||
let db = engine.db.lock().unwrap();
|
||||
assert!(db.get_node(node_id).unwrap().is_none());
|
||||
}
|
||||
|
||||
// Roll back the rollback (re-create)
|
||||
let _rr = engine.rollback_rollback(rb.id).unwrap();
|
||||
|
||||
// Node should be back
|
||||
let db = engine.db.lock().unwrap();
|
||||
let found = db.get_node(node_id).unwrap();
|
||||
assert!(found.is_some());
|
||||
assert_eq!(found.unwrap().content, b"test content");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_history() {
|
||||
let (mut engine, _db_dir, _log_dir) = make_engine();
|
||||
let node = make_node();
|
||||
let cmd = build_create_node_cmd(&node, "hist-1", None);
|
||||
engine.apply(cmd).unwrap();
|
||||
|
||||
let history = engine.history(0).unwrap();
|
||||
assert!(!history.is_empty());
|
||||
assert!(history.iter().any(|c| matches!(c.command_type, CommandType::CreateNode)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_causal_chain() {
|
||||
let (mut engine, _db_dir, _log_dir) = make_engine();
|
||||
let node = make_node();
|
||||
let cmd = build_create_node_cmd(&node, "causal-1", None);
|
||||
let result = engine.apply(cmd).unwrap();
|
||||
|
||||
// Roll back (creates a child command with causal_parent = create_cmd.id)
|
||||
let rb = engine.rollback(result.command_id).unwrap();
|
||||
|
||||
let chain = engine.causal_chain(rb.id).unwrap();
|
||||
// Chain should be [create_cmd, rollback_cmd]
|
||||
assert_eq!(chain.len(), 2);
|
||||
assert!(matches!(chain[0].command_type, CommandType::CreateNode));
|
||||
assert!(matches!(chain[1].command_type, CommandType::Rollback(_)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_create_edge_command() {
|
||||
let (mut engine, _db_dir, _log_dir) = make_engine();
|
||||
|
||||
// Create two nodes first
|
||||
let n1 = make_node();
|
||||
let n2 = Node::new(
|
||||
NodeType::Concept,
|
||||
vec![0.0, 1.0],
|
||||
b"concept".to_vec(),
|
||||
MemoryTier::Semantic,
|
||||
0.5,
|
||||
);
|
||||
engine.apply(build_create_node_cmd(&n1, "edge-n1", None)).unwrap();
|
||||
engine.apply(build_create_node_cmd(&n2, "edge-n2", None)).unwrap();
|
||||
|
||||
// Create edge
|
||||
let edge = Edge::new(n1.id, n2.id, RelationType::References, 0.7);
|
||||
let edge_cmd = build_create_edge_cmd(&edge, "edge-create-1", None);
|
||||
let result = engine.apply(edge_cmd).unwrap();
|
||||
assert_eq!(result.status, CommandStatus::Applied);
|
||||
|
||||
// Verify edge exists
|
||||
let db = engine.db.lock().unwrap();
|
||||
let edges = db.get_edges_from(n1.id).unwrap();
|
||||
assert!(!edges.is_empty());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
use thiserror::Error;
|
||||
use uuid::Uuid;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum TxError {
|
||||
#[error("Command not found: {0}")]
|
||||
NotFound(Uuid),
|
||||
|
||||
#[error("Command already applied (idempotency key: {0})")]
|
||||
AlreadyApplied(String),
|
||||
|
||||
#[error("Cannot roll back a command in status {0:?}")]
|
||||
InvalidStatus(String),
|
||||
|
||||
#[error("Conflict: {0}")]
|
||||
Conflict(String),
|
||||
|
||||
#[error("JSON error: {0}")]
|
||||
Json(#[from] serde_json::Error),
|
||||
|
||||
#[error("Storage error: {0}")]
|
||||
Storage(#[from] sled::Error),
|
||||
|
||||
#[error("Engram error: {0}")]
|
||||
Engram(#[from] engram_core::EngramError),
|
||||
|
||||
#[error("Invalid command: {0}")]
|
||||
Invalid(String),
|
||||
}
|
||||
|
||||
pub type TxResult<T> = Result<T, TxError>;
|
||||
@@ -0,0 +1,29 @@
|
||||
/// Engram Transaction Engine — Command pattern with rollback-of-rollback.
|
||||
///
|
||||
/// # The Central Insight
|
||||
///
|
||||
/// Every mutation to Engram is a Command — a first-class object with:
|
||||
/// - The operation and its inverse (computed eagerly at command time)
|
||||
/// - An idempotency key (same key = same command, applied once)
|
||||
/// - A causal parent (which command caused this one)
|
||||
/// - A timestamp and originating peer ID
|
||||
///
|
||||
/// Commands form a DAG of causality. You can roll back any command. You can
|
||||
/// roll back a rollback (undo an undo — re-applying the original). The
|
||||
/// command log is append-only: history is never rewritten.
|
||||
///
|
||||
/// # Rollback-of-Rollback
|
||||
///
|
||||
/// When you roll back command X, a new `Rollback(X)` command is created and
|
||||
/// applied. Its `inverse_payload` is X's original `payload`. If you then roll
|
||||
/// back the rollback, a new `Rollback(rollback_id)` is created, whose effect
|
||||
/// is to re-apply X. This is a full undo/redo system with causal lineage.
|
||||
pub mod command;
|
||||
pub mod engine;
|
||||
pub mod error;
|
||||
pub mod log;
|
||||
|
||||
pub use command::{Command, CommandResult, CommandStatus, CommandType};
|
||||
pub use engine::TransactionEngine;
|
||||
pub use error::TxError;
|
||||
pub use log::CommandLog;
|
||||
@@ -0,0 +1,151 @@
|
||||
/// CommandLog — append-only log of all commands, stored in sled.
|
||||
///
|
||||
/// Key schema:
|
||||
/// cmd:{uuid} → JSON-encoded Command (JSON used because Command contains serde_json::Value)
|
||||
/// idem:{key} → uuid bytes (idempotency index)
|
||||
/// cmd_ts:{ms}:{uuid} → uuid bytes (time-ordered scan index)
|
||||
use sled::Db;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::command::Command;
|
||||
use crate::error::{TxError, TxResult};
|
||||
|
||||
pub struct CommandLog {
|
||||
db: Db,
|
||||
}
|
||||
|
||||
impl CommandLog {
|
||||
pub fn open(db: Db) -> Self {
|
||||
Self { db }
|
||||
}
|
||||
|
||||
// ── Write ─────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Append a command to the log. Overwrites if the UUID already exists
|
||||
/// (used for status updates after application).
|
||||
pub fn write(&self, cmd: &Command) -> TxResult<()> {
|
||||
let key = cmd_key(cmd.id);
|
||||
// Use JSON (not bincode) because Command contains serde_json::Value,
|
||||
// which bincode cannot deserialize (DeserializeAnyNotSupported).
|
||||
let val = serde_json::to_vec(cmd)?;
|
||||
self.db.insert(key, val)?;
|
||||
|
||||
// Idempotency index: idem:{key} → uuid
|
||||
let idem_key = idem_key(&cmd.idempotency_key);
|
||||
self.db.insert(idem_key, cmd.id.as_bytes().to_vec())?;
|
||||
|
||||
// Time index: cmd_ts:{ms:016x}:{uuid} → uuid
|
||||
let ts_key = ts_key(cmd.timestamp_ms, cmd.id);
|
||||
self.db.insert(ts_key, cmd.id.as_bytes().to_vec())?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ── Read ──────────────────────────────────────────────────────────────────
|
||||
|
||||
pub fn get(&self, id: Uuid) -> TxResult<Option<Command>> {
|
||||
match self.db.get(cmd_key(id))? {
|
||||
Some(bytes) => Ok(Some(serde_json::from_slice(&bytes)?)),
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn require(&self, id: Uuid) -> TxResult<Command> {
|
||||
self.get(id)?.ok_or(TxError::NotFound(id))
|
||||
}
|
||||
|
||||
/// Check whether an idempotency key has already been applied.
|
||||
/// Returns the command UUID if it exists.
|
||||
pub fn check_idempotency(&self, key: &str) -> TxResult<Option<Uuid>> {
|
||||
match self.db.get(idem_key(key))? {
|
||||
Some(bytes) => {
|
||||
let arr: [u8; 16] = bytes[..16]
|
||||
.try_into()
|
||||
.map_err(|_| TxError::Invalid("bad uuid bytes in idem index".into()))?;
|
||||
Ok(Some(Uuid::from_bytes(arr)))
|
||||
}
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
/// Load all commands created at or after `since_ms`, ordered by timestamp.
|
||||
pub fn since(&self, since_ms: i64) -> TxResult<Vec<Command>> {
|
||||
let prefix = format!("cmd_ts:{:016x}:", since_ms);
|
||||
let mut cmds = Vec::new();
|
||||
for result in self.db.range(prefix.as_bytes()..) {
|
||||
let (k, _v) = result?;
|
||||
// Check the key starts with "cmd_ts:"
|
||||
if !k.starts_with(b"cmd_ts:") {
|
||||
break;
|
||||
}
|
||||
// Extract uuid from key: cmd_ts:{ms}:{uuid}
|
||||
let key_str = std::str::from_utf8(&k)
|
||||
.map_err(|e| TxError::Invalid(e.to_string()))?;
|
||||
let parts: Vec<&str> = key_str.splitn(3, ':').collect();
|
||||
if parts.len() < 3 {
|
||||
continue;
|
||||
}
|
||||
// parts[1] = ms (hex), parts[2] = uuid
|
||||
let ts_hex = parts[1];
|
||||
let ts = i64::from_str_radix(ts_hex, 16)
|
||||
.map_err(|e| TxError::Invalid(e.to_string()))?;
|
||||
if ts < since_ms {
|
||||
continue;
|
||||
}
|
||||
let id: Uuid = parts[2]
|
||||
.parse()
|
||||
.map_err(|e: uuid::Error| TxError::Invalid(e.to_string()))?;
|
||||
if let Some(cmd) = self.get(id)? {
|
||||
cmds.push(cmd);
|
||||
}
|
||||
}
|
||||
Ok(cmds)
|
||||
}
|
||||
|
||||
/// Load all commands in the store.
|
||||
pub fn all(&self) -> TxResult<Vec<Command>> {
|
||||
self.since(0)
|
||||
}
|
||||
|
||||
/// Collect the causal chain leading to a given command (inclusive).
|
||||
/// Walks `causal_parent` links back to the root.
|
||||
pub fn causal_chain(&self, id: Uuid) -> TxResult<Vec<Command>> {
|
||||
let mut chain = Vec::new();
|
||||
let mut current_id = Some(id);
|
||||
let mut visited = std::collections::HashSet::new();
|
||||
|
||||
while let Some(cid) = current_id {
|
||||
if visited.contains(&cid) {
|
||||
break; // cycle guard
|
||||
}
|
||||
visited.insert(cid);
|
||||
|
||||
match self.get(cid)? {
|
||||
Some(cmd) => {
|
||||
current_id = cmd.causal_parent;
|
||||
chain.push(cmd);
|
||||
}
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
|
||||
// Return root-first (reverse of walk order)
|
||||
chain.reverse();
|
||||
Ok(chain)
|
||||
}
|
||||
}
|
||||
|
||||
// ── Key constructors ──────────────────────────────────────────────────────────
|
||||
|
||||
fn cmd_key(id: Uuid) -> Vec<u8> {
|
||||
format!("cmd:{}", id).into_bytes()
|
||||
}
|
||||
|
||||
fn idem_key(key: &str) -> Vec<u8> {
|
||||
format!("idem:{}", key).into_bytes()
|
||||
}
|
||||
|
||||
fn ts_key(ts: i64, id: Uuid) -> Vec<u8> {
|
||||
// Zero-padded hex timestamp for lexicographic ordering
|
||||
format!("cmd_ts:{:016x}:{}", ts, id).into_bytes()
|
||||
}
|
||||
Reference in New Issue
Block a user