use crate::{
Config, Error, MergePair, Node, NodeData, NodeHistory, NodeId, ObjectId, ObjectPayload, Owner,
Provenance, Result, TransactionId, TransactionPackage, WriterId,
graph::{Graph, Record},
ledger, nodes, objects,
projector::Projection,
wire::{self, NodeOperation, ObjectDeclaration, ParsedTransaction, UnsignedTransaction},
};
use chrono::Utc;
use ed25519_dalek::SigningKey;
use fs2::FileExt;
use std::{
collections::{BTreeMap, BTreeSet},
fmt,
fs::{self, File, OpenOptions},
path::{Path, PathBuf},
sync::{Arc, Mutex, MutexGuard, RwLock},
};
#[derive(Clone)]
pub struct KwebDb {
inner: Arc<Inner>,
}
struct Inner {
root: PathBuf,
_lock_file: File,
mutation: Mutex<()>,
state: RwLock<DbState>,
signing_key: SigningKey,
local_writer: WriterId,
writers: Vec<WriterId>,
gossip: Arc<dyn crate::Gossip>,
}
#[derive(Clone)]
struct DbState {
graph: Graph,
projection: Projection,
}
impl fmt::Debug for KwebDb {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("KwebDb")
.field("root", &self.inner.root)
.finish_non_exhaustive()
}
}
impl KwebDb {
pub fn open(path: impl AsRef<Path>, config: Config) -> Result<Self> {
validate_config(&config)?;
let root = path.as_ref().to_path_buf();
match fs::symlink_metadata(&root) {
Ok(metadata) => {
if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
return Err(Error::invalid_config(
"database root must be a real directory",
));
}
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
fs::create_dir_all(&root)?;
}
Err(error) => return Err(error.into()),
}
let metadata = fs::symlink_metadata(&root)?;
if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
return Err(Error::invalid_config(
"database root must be a real directory",
));
}
let lock_file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(root.join("LOCK"))?;
FileExt::try_lock_exclusive(&lock_file).map_err(|error| {
Error::Busy(format!(
"cannot exclusively lock {}: {error}",
root.display()
))
})?;
ledger::initialize(&root)?;
objects::initialize(&root)?;
let mut graph = Graph::default();
for transaction in ledger::read_frames(&root)? {
let parsed = wire::parse_signed(&transaction)?;
ensure_authorized(parsed.unsigned.writer, &config.writers_by_priority)?;
let package = objects::load_package(&root, transaction, &parsed)?;
objects::verify_package(&parsed, &package)?;
let _ = graph.insert(Record { parsed, package })?;
}
let projection = Projection::build(&graph, &config.writers_by_priority);
nodes::rebuild(&root, &projection)?;
let signing_key = SigningKey::from_bytes(&config.signing_key);
let local_writer = WriterId(signing_key.verifying_key().to_bytes());
let database = Self {
inner: Arc::new(Inner {
root,
_lock_file: lock_file,
mutation: Mutex::new(()),
state: RwLock::new(DbState { graph, projection }),
signing_key,
local_writer,
writers: config.writers_by_priority,
gossip: config.gossip,
}),
};
database.reannounce();
Ok(database)
}
pub fn start_transaction(&self, provenance: Provenance) -> Result<Transaction<'_>> {
provenance.validate()?;
let guard = self
.inner
.mutation
.lock()
.map_err(|_| Error::corrupt("mutation mutex is poisoned"))?;
let state = self
.inner
.state
.read()
.map_err(|_| Error::corrupt("projection lock is poisoned"))?;
let transaction = Transaction {
db: self,
guard: Some(guard),
provenance,
heads: state.graph.heads(),
known_nodes: state.graph.known_node_ids(),
known_objects: state.graph.known_object_ids(),
objects: BTreeMap::new(),
creates: BTreeMap::new(),
updates: BTreeMap::new(),
connections: BTreeSet::new(),
merges: BTreeSet::new(),
};
Ok(transaction)
}
pub fn accept_transaction(&self, package: TransactionPackage) -> Result<bool> {
let guard = self
.inner
.mutation
.lock()
.map_err(|_| Error::corrupt("mutation mutex is poisoned"))?;
let parsed = wire::parse_signed(&package.transaction)?;
ensure_authorized(parsed.unsigned.writer, &self.inner.writers)?;
objects::verify_package(&parsed, &package)?;
{
let state = self
.inner
.state
.read()
.map_err(|_| Error::corrupt("projection lock is poisoned"))?;
if let Some(existing) = state.graph.get(parsed.id) {
if existing.package == package {
return Ok(false);
}
return Err(Error::corrupt(
"retained transaction ID has a different package",
));
}
}
self.commit_new(parsed, package.clone())?;
drop(guard);
self.inner.gossip.announce(package);
Ok(true)
}
pub fn get_node(&self, id: NodeId) -> Result<Node> {
let state = self
.inner
.state
.read()
.map_err(|_| Error::corrupt("projection lock is poisoned"))?;
state
.projection
.nodes
.get(&id)
.map(|projected| projected.node.clone())
.ok_or_else(|| Error::not_found(format!("node {id}")))
}
pub fn get_node_history(&self, id: NodeId) -> Result<NodeHistory> {
let state = self
.inner
.state
.read()
.map_err(|_| Error::corrupt("projection lock is poisoned"))?;
state
.projection
.histories
.get(&id)
.cloned()
.ok_or_else(|| Error::not_found(format!("node history {id}")))
}
pub fn get_object(&self, id: ObjectId) -> Result<Vec<u8>> {
let transaction = {
let state = self
.inner
.state
.read()
.map_err(|_| Error::corrupt("projection lock is poisoned"))?;
state
.projection
.objects
.get(&id)
.map(|object| object.visible_transaction)
.ok_or_else(|| Error::not_found(format!("object {id}")))?
};
objects::read_visible(&self.inner.root, id, transaction)
}
fn commit_new(&self, parsed: ParsedTransaction, package: TransactionPackage) -> Result<()> {
let (graph, projection) = {
let state = self
.inner
.state
.read()
.map_err(|_| Error::corrupt("projection lock is poisoned"))?;
let mut graph = state.graph.clone();
if !graph.insert(Record {
parsed: parsed.clone(),
package: package.clone(),
})? {
return Err(Error::corrupt(
"new-transaction commit encountered a duplicate",
));
}
let projection = Projection::build(&graph, &self.inner.writers);
(graph, projection)
};
objects::install_package(&self.inner.root, &parsed, &package)?;
nodes::write_projection(&self.inner.root, &projection)?;
ledger::append(&self.inner.root, parsed.id, &package.transaction)?;
let mut state = self
.inner
.state
.write()
.map_err(|_| Error::corrupt("projection lock is poisoned"))?;
*state = DbState { graph, projection };
Ok(())
}
fn reannounce(&self) {
let packages = {
let state = self
.inner
.state
.read()
.expect("projection lock cannot be poisoned during open");
state
.graph
.records()
.map(|(_, record)| record.package.clone())
.collect::<Vec<_>>()
};
for package in packages {
self.inner.gossip.announce(package);
}
}
}
pub struct Transaction<'a> {
db: &'a KwebDb,
guard: Option<MutexGuard<'a, ()>>,
provenance: Provenance,
heads: Vec<TransactionId>,
known_nodes: BTreeSet<NodeId>,
known_objects: BTreeSet<ObjectId>,
objects: BTreeMap<ObjectId, Vec<u8>>,
creates: BTreeMap<NodeId, NodeData>,
updates: BTreeMap<NodeId, NodeData>,
connections: BTreeSet<(NodeId, NodeId)>,
merges: BTreeSet<MergePair>,
}
impl fmt::Debug for Transaction<'_> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("Transaction")
.field("heads", &self.heads)
.field("objects", &self.objects.len())
.field("creates", &self.creates.len())
.field("updates", &self.updates.len())
.field("connections", &self.connections.len())
.field("merges", &self.merges.len())
.finish_non_exhaustive()
}
}
impl Transaction<'_> {
pub fn create_object(&mut self, bytes: Vec<u8>) -> Result<ObjectId> {
if bytes.len() > crate::model::MAX_OBJECT_BYTES {
return Err(Error::invalid_input("object exceeds the 64 MiB limit"));
}
let id = loop {
let candidate = ObjectId::random();
if !self.known_objects.contains(&candidate) && !self.objects.contains_key(&candidate) {
break candidate;
}
};
self.objects.insert(id, bytes);
Ok(id)
}
pub fn create_node(&mut self, data: NodeData) -> Result<NodeId> {
data.validate()?;
let id = loop {
let candidate = NodeId::random();
if !self.known_nodes.contains(&candidate) && !self.creates.contains_key(&candidate) {
break candidate;
}
};
self.creates.insert(id, data);
Ok(id)
}
pub fn update_node(&mut self, id: NodeId, data: NodeData) -> Result<()> {
if self.creates.contains_key(&id) {
return Err(Error::invalid_input(
"one transaction cannot create and update the same node",
));
}
data.validate()?;
if self.updates.insert(id, data).is_some() {
return Err(Error::invalid_input(
"one transaction cannot update the same node twice",
));
}
Ok(())
}
pub fn connect_node(&mut self, from: NodeId, to: NodeId) -> Result<()> {
self.connections.insert((from, to));
Ok(())
}
pub fn merge(&mut self, first: TransactionId, second: TransactionId) -> Result<()> {
let pair = MergePair::new(first, second)?;
if !self.merges.insert(pair) {
return Err(Error::invalid_input("duplicate merge pair"));
}
Ok(())
}
pub fn finalize(mut self) -> Result<TransactionId> {
self.validate_local_projection()?;
let unsigned = UnsignedTransaction {
writer: self.db.inner.local_writer,
committed_at: Utc::now(),
heads: self.heads.clone(),
provenance: self.provenance.clone(),
merge_pairs: self.merges.iter().copied().collect(),
objects: self
.objects
.iter()
.map(|(id, bytes)| ObjectDeclaration {
id: *id,
length: bytes.len() as u64,
sha256: wire::object_hash(bytes),
})
.collect(),
creates: self
.creates
.iter()
.map(|(id, data)| NodeOperation {
id: *id,
data: data.clone(),
})
.collect(),
updates: self
.updates
.iter()
.map(|(id, data)| NodeOperation {
id: *id,
data: data.clone(),
})
.collect(),
connections: self.connections.iter().copied().collect(),
};
let transaction = wire::build_signed(&unsigned, &self.db.inner.signing_key)?;
let parsed = wire::parse_signed(&transaction)?;
let package = TransactionPackage {
transaction,
objects: self
.objects
.into_iter()
.map(|(id, bytes)| ObjectPayload { id, bytes })
.collect(),
};
objects::verify_package(&parsed, &package)?;
self.db.commit_new(parsed.clone(), package.clone())?;
drop(self.guard.take());
self.db.inner.gossip.announce(package);
Ok(parsed.id)
}
fn validate_local_projection(&self) -> Result<()> {
let state = self
.db
.inner
.state
.read()
.map_err(|_| Error::corrupt("projection lock is poisoned"))?;
let mut nodes = state
.projection
.nodes
.keys()
.copied()
.collect::<BTreeSet<_>>();
nodes.extend(self.creates.keys().copied());
let mut objects = state
.projection
.objects
.keys()
.copied()
.collect::<BTreeSet<_>>();
objects.extend(self.objects.keys().copied());
for (id, data) in &self.creates {
validate_references(*id, data, &nodes, &objects)?;
}
for (id, data) in &self.updates {
if !state.projection.node_exists(*id) {
return Err(Error::invalid_input(format!(
"cannot update nonvisible node {id}"
)));
}
validate_references(*id, data, &nodes, &objects)?;
}
for (from, to) in &self.connections {
if !nodes.contains(from) || !nodes.contains(to) {
return Err(Error::invalid_input(
"connection endpoints must be locally resolvable",
));
}
}
for pair in &self.merges {
let targets = state.projection.pair_update_targets(*pair);
if targets.is_empty() {
return Err(Error::invalid_input(
"merge pair is not an exact current shared node frontier",
));
}
if !targets.iter().any(|node| self.updates.contains_key(node)) {
return Err(Error::invalid_input(
"merge pair requires an update for a node it conflicts on",
));
}
}
Ok(())
}
}
fn validate_config(config: &Config) -> Result<()> {
if config.writers_by_priority.is_empty() {
return Err(Error::invalid_config("writers_by_priority cannot be empty"));
}
let writers = config
.writers_by_priority
.iter()
.copied()
.collect::<BTreeSet<_>>();
if writers.len() != config.writers_by_priority.len() {
return Err(Error::invalid_config("writers_by_priority must be unique"));
}
if !writers.contains(&WriterId::from_signing_key(&config.signing_key)) {
return Err(Error::invalid_config(
"writers_by_priority must contain the local writer",
));
}
Ok(())
}
fn ensure_authorized(writer: WriterId, writers: &[WriterId]) -> Result<()> {
if writers.contains(&writer) {
Ok(())
} else {
Err(Error::invalid_transaction(format!(
"writer {writer} is not configured"
)))
}
}
fn validate_references(
self_id: NodeId,
data: &NodeData,
nodes: &BTreeSet<NodeId>,
objects: &BTreeSet<ObjectId>,
) -> Result<()> {
if let Owner::Node(owner) = data.owner
&& !nodes.contains(&owner)
&& owner != self_id
{
return Err(Error::invalid_input(format!(
"owner node {owner} is not locally resolvable"
)));
}
for reference in data.fixed_connections.iter().flatten() {
if !nodes.contains(reference) && *reference != self_id {
return Err(Error::invalid_input(format!(
"fixed connection {reference} is not locally resolvable"
)));
}
}
for reference in &data.objects {
if !objects.contains(reference) {
return Err(Error::invalid_input(format!(
"object {reference} is not locally resolvable"
)));
}
}
Ok(())
}