1use crate::{
2 Config, Error, MergePair, Node, NodeData, NodeHistory, NodeId, ObjectId, ObjectPayload, Owner,
3 Provenance, Result, TransactionId, TransactionPackage, WriterId,
4 graph::{Graph, Record},
5 ledger, nodes, objects,
6 projector::Projection,
7 wire::{self, NodeOperation, ObjectDeclaration, ParsedTransaction, UnsignedTransaction},
8};
9use chrono::Utc;
10use ed25519_dalek::SigningKey;
11use fs2::FileExt;
12use std::{
13 collections::{BTreeMap, BTreeSet},
14 fmt,
15 fs::{self, File, OpenOptions},
16 path::{Path, PathBuf},
17 sync::{Arc, Mutex, MutexGuard, RwLock},
18};
19
20#[derive(Clone)]
21pub struct KwebDb {
22 inner: Arc<Inner>,
23}
24
25struct Inner {
26 root: PathBuf,
27 _lock_file: File,
28 mutation: Mutex<()>,
29 state: RwLock<DbState>,
30 signing_key: SigningKey,
31 local_writer: WriterId,
32 writers: Vec<WriterId>,
33 gossip: Arc<dyn crate::Gossip>,
34}
35
36#[derive(Clone)]
37struct DbState {
38 graph: Graph,
39 projection: Projection,
40}
41
42impl fmt::Debug for KwebDb {
43 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
44 formatter
45 .debug_struct("KwebDb")
46 .field("root", &self.inner.root)
47 .finish_non_exhaustive()
48 }
49}
50
51impl KwebDb {
52 pub fn open(path: impl AsRef<Path>, config: Config) -> Result<Self> {
53 validate_config(&config)?;
54 let root = path.as_ref().to_path_buf();
55 match fs::symlink_metadata(&root) {
56 Ok(metadata) => {
57 if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
58 return Err(Error::invalid_config(
59 "database root must be a real directory",
60 ));
61 }
62 }
63 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
64 fs::create_dir_all(&root)?;
65 }
66 Err(error) => return Err(error.into()),
67 }
68 let metadata = fs::symlink_metadata(&root)?;
69 if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
70 return Err(Error::invalid_config(
71 "database root must be a real directory",
72 ));
73 }
74 let lock_file = OpenOptions::new()
75 .read(true)
76 .write(true)
77 .create(true)
78 .truncate(false)
79 .open(root.join("LOCK"))?;
80 FileExt::try_lock_exclusive(&lock_file).map_err(|error| {
81 Error::Busy(format!(
82 "cannot exclusively lock {}: {error}",
83 root.display()
84 ))
85 })?;
86 ledger::initialize(&root)?;
87 objects::initialize(&root)?;
88
89 let mut graph = Graph::default();
90 for transaction in ledger::read_frames(&root)? {
91 let parsed = wire::parse_signed(&transaction)?;
92 ensure_authorized(parsed.unsigned.writer, &config.writers_by_priority)?;
93 let package = objects::load_package(&root, transaction, &parsed)?;
94 objects::verify_package(&parsed, &package)?;
95 let _ = graph.insert(Record { parsed, package })?;
96 }
97 let projection = Projection::build(&graph, &config.writers_by_priority);
98 nodes::rebuild(&root, &projection)?;
99 let signing_key = SigningKey::from_bytes(&config.signing_key);
100 let local_writer = WriterId(signing_key.verifying_key().to_bytes());
101 let database = Self {
102 inner: Arc::new(Inner {
103 root,
104 _lock_file: lock_file,
105 mutation: Mutex::new(()),
106 state: RwLock::new(DbState { graph, projection }),
107 signing_key,
108 local_writer,
109 writers: config.writers_by_priority,
110 gossip: config.gossip,
111 }),
112 };
113 database.reannounce();
114 Ok(database)
115 }
116
117 pub fn start_transaction(&self, provenance: Provenance) -> Result<Transaction<'_>> {
118 provenance.validate()?;
119 let guard = self
120 .inner
121 .mutation
122 .lock()
123 .map_err(|_| Error::corrupt("mutation mutex is poisoned"))?;
124 let state = self
125 .inner
126 .state
127 .read()
128 .map_err(|_| Error::corrupt("projection lock is poisoned"))?;
129 let transaction = Transaction {
130 db: self,
131 guard: Some(guard),
132 provenance,
133 heads: state.graph.heads(),
134 known_nodes: state.graph.known_node_ids(),
135 known_objects: state.graph.known_object_ids(),
136 objects: BTreeMap::new(),
137 creates: BTreeMap::new(),
138 updates: BTreeMap::new(),
139 connections: BTreeSet::new(),
140 merges: BTreeSet::new(),
141 };
142 Ok(transaction)
143 }
144
145 pub fn accept_transaction(&self, package: TransactionPackage) -> Result<bool> {
146 let guard = self
147 .inner
148 .mutation
149 .lock()
150 .map_err(|_| Error::corrupt("mutation mutex is poisoned"))?;
151 let parsed = wire::parse_signed(&package.transaction)?;
152 ensure_authorized(parsed.unsigned.writer, &self.inner.writers)?;
153 objects::verify_package(&parsed, &package)?;
154 {
155 let state = self
156 .inner
157 .state
158 .read()
159 .map_err(|_| Error::corrupt("projection lock is poisoned"))?;
160 if let Some(existing) = state.graph.get(parsed.id) {
161 if existing.package == package {
162 return Ok(false);
163 }
164 return Err(Error::corrupt(
165 "retained transaction ID has a different package",
166 ));
167 }
168 }
169 self.commit_new(parsed, package.clone())?;
170 drop(guard);
171 self.inner.gossip.announce(package);
172 Ok(true)
173 }
174
175 pub fn get_node(&self, id: NodeId) -> Result<Node> {
176 let state = self
177 .inner
178 .state
179 .read()
180 .map_err(|_| Error::corrupt("projection lock is poisoned"))?;
181 state
182 .projection
183 .nodes
184 .get(&id)
185 .map(|projected| projected.node.clone())
186 .ok_or_else(|| Error::not_found(format!("node {id}")))
187 }
188
189 pub fn get_node_history(&self, id: NodeId) -> Result<NodeHistory> {
190 let state = self
191 .inner
192 .state
193 .read()
194 .map_err(|_| Error::corrupt("projection lock is poisoned"))?;
195 state
196 .projection
197 .histories
198 .get(&id)
199 .cloned()
200 .ok_or_else(|| Error::not_found(format!("node history {id}")))
201 }
202
203 pub fn get_object(&self, id: ObjectId) -> Result<Vec<u8>> {
204 let transaction = {
205 let state = self
206 .inner
207 .state
208 .read()
209 .map_err(|_| Error::corrupt("projection lock is poisoned"))?;
210 state
211 .projection
212 .objects
213 .get(&id)
214 .map(|object| object.visible_transaction)
215 .ok_or_else(|| Error::not_found(format!("object {id}")))?
216 };
217 objects::read_visible(&self.inner.root, id, transaction)
218 }
219
220 fn commit_new(&self, parsed: ParsedTransaction, package: TransactionPackage) -> Result<()> {
221 let (graph, projection) = {
222 let state = self
223 .inner
224 .state
225 .read()
226 .map_err(|_| Error::corrupt("projection lock is poisoned"))?;
227 let mut graph = state.graph.clone();
228 if !graph.insert(Record {
229 parsed: parsed.clone(),
230 package: package.clone(),
231 })? {
232 return Err(Error::corrupt(
233 "new-transaction commit encountered a duplicate",
234 ));
235 }
236 let projection = Projection::build(&graph, &self.inner.writers);
237 (graph, projection)
238 };
239 objects::install_package(&self.inner.root, &parsed, &package)?;
240 nodes::write_projection(&self.inner.root, &projection)?;
241 ledger::append(&self.inner.root, parsed.id, &package.transaction)?;
242 let mut state = self
243 .inner
244 .state
245 .write()
246 .map_err(|_| Error::corrupt("projection lock is poisoned"))?;
247 *state = DbState { graph, projection };
248 Ok(())
249 }
250
251 fn reannounce(&self) {
252 let packages = {
253 let state = self
254 .inner
255 .state
256 .read()
257 .expect("projection lock cannot be poisoned during open");
258 state
259 .graph
260 .records()
261 .map(|(_, record)| record.package.clone())
262 .collect::<Vec<_>>()
263 };
264 for package in packages {
265 self.inner.gossip.announce(package);
266 }
267 }
268}
269
270pub struct Transaction<'a> {
271 db: &'a KwebDb,
272 guard: Option<MutexGuard<'a, ()>>,
273 provenance: Provenance,
274 heads: Vec<TransactionId>,
275 known_nodes: BTreeSet<NodeId>,
276 known_objects: BTreeSet<ObjectId>,
277 objects: BTreeMap<ObjectId, Vec<u8>>,
278 creates: BTreeMap<NodeId, NodeData>,
279 updates: BTreeMap<NodeId, NodeData>,
280 connections: BTreeSet<(NodeId, NodeId)>,
281 merges: BTreeSet<MergePair>,
282}
283
284impl fmt::Debug for Transaction<'_> {
285 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
286 formatter
287 .debug_struct("Transaction")
288 .field("heads", &self.heads)
289 .field("objects", &self.objects.len())
290 .field("creates", &self.creates.len())
291 .field("updates", &self.updates.len())
292 .field("connections", &self.connections.len())
293 .field("merges", &self.merges.len())
294 .finish_non_exhaustive()
295 }
296}
297
298impl Transaction<'_> {
299 pub fn create_object(&mut self, bytes: Vec<u8>) -> Result<ObjectId> {
300 if bytes.len() > crate::model::MAX_OBJECT_BYTES {
301 return Err(Error::invalid_input("object exceeds the 64 MiB limit"));
302 }
303 let id = loop {
304 let candidate = ObjectId::random();
305 if !self.known_objects.contains(&candidate) && !self.objects.contains_key(&candidate) {
306 break candidate;
307 }
308 };
309 self.objects.insert(id, bytes);
310 Ok(id)
311 }
312
313 pub fn create_node(&mut self, data: NodeData) -> Result<NodeId> {
314 data.validate()?;
315 let id = loop {
316 let candidate = NodeId::random();
317 if !self.known_nodes.contains(&candidate) && !self.creates.contains_key(&candidate) {
318 break candidate;
319 }
320 };
321 self.creates.insert(id, data);
322 Ok(id)
323 }
324
325 pub fn update_node(&mut self, id: NodeId, data: NodeData) -> Result<()> {
326 if self.creates.contains_key(&id) {
327 return Err(Error::invalid_input(
328 "one transaction cannot create and update the same node",
329 ));
330 }
331 data.validate()?;
332 if self.updates.insert(id, data).is_some() {
333 return Err(Error::invalid_input(
334 "one transaction cannot update the same node twice",
335 ));
336 }
337 Ok(())
338 }
339
340 pub fn connect_node(&mut self, from: NodeId, to: NodeId) -> Result<()> {
341 self.connections.insert((from, to));
342 Ok(())
343 }
344
345 pub fn merge(&mut self, first: TransactionId, second: TransactionId) -> Result<()> {
346 let pair = MergePair::new(first, second)?;
347 if !self.merges.insert(pair) {
348 return Err(Error::invalid_input("duplicate merge pair"));
349 }
350 Ok(())
351 }
352
353 pub fn finalize(mut self) -> Result<TransactionId> {
354 self.validate_local_projection()?;
355 let unsigned = UnsignedTransaction {
356 writer: self.db.inner.local_writer,
357 committed_at: Utc::now(),
358 heads: self.heads.clone(),
359 provenance: self.provenance.clone(),
360 merge_pairs: self.merges.iter().copied().collect(),
361 objects: self
362 .objects
363 .iter()
364 .map(|(id, bytes)| ObjectDeclaration {
365 id: *id,
366 length: bytes.len() as u64,
367 sha256: wire::object_hash(bytes),
368 })
369 .collect(),
370 creates: self
371 .creates
372 .iter()
373 .map(|(id, data)| NodeOperation {
374 id: *id,
375 data: data.clone(),
376 })
377 .collect(),
378 updates: self
379 .updates
380 .iter()
381 .map(|(id, data)| NodeOperation {
382 id: *id,
383 data: data.clone(),
384 })
385 .collect(),
386 connections: self.connections.iter().copied().collect(),
387 };
388 let transaction = wire::build_signed(&unsigned, &self.db.inner.signing_key)?;
389 let parsed = wire::parse_signed(&transaction)?;
390 let package = TransactionPackage {
391 transaction,
392 objects: self
393 .objects
394 .into_iter()
395 .map(|(id, bytes)| ObjectPayload { id, bytes })
396 .collect(),
397 };
398 objects::verify_package(&parsed, &package)?;
399 self.db.commit_new(parsed.clone(), package.clone())?;
400 drop(self.guard.take());
401 self.db.inner.gossip.announce(package);
402 Ok(parsed.id)
403 }
404
405 fn validate_local_projection(&self) -> Result<()> {
406 let state = self
407 .db
408 .inner
409 .state
410 .read()
411 .map_err(|_| Error::corrupt("projection lock is poisoned"))?;
412 let mut nodes = state
413 .projection
414 .nodes
415 .keys()
416 .copied()
417 .collect::<BTreeSet<_>>();
418 nodes.extend(self.creates.keys().copied());
419 let mut objects = state
420 .projection
421 .objects
422 .keys()
423 .copied()
424 .collect::<BTreeSet<_>>();
425 objects.extend(self.objects.keys().copied());
426 for (id, data) in &self.creates {
427 validate_references(*id, data, &nodes, &objects)?;
428 }
429 for (id, data) in &self.updates {
430 if !state.projection.node_exists(*id) {
431 return Err(Error::invalid_input(format!(
432 "cannot update nonvisible node {id}"
433 )));
434 }
435 validate_references(*id, data, &nodes, &objects)?;
436 }
437 for (from, to) in &self.connections {
438 if !nodes.contains(from) || !nodes.contains(to) {
439 return Err(Error::invalid_input(
440 "connection endpoints must be locally resolvable",
441 ));
442 }
443 }
444 for pair in &self.merges {
445 let targets = state.projection.pair_update_targets(*pair);
446 if targets.is_empty() {
447 return Err(Error::invalid_input(
448 "merge pair is not an exact current shared node frontier",
449 ));
450 }
451 if !targets.iter().any(|node| self.updates.contains_key(node)) {
452 return Err(Error::invalid_input(
453 "merge pair requires an update for a node it conflicts on",
454 ));
455 }
456 }
457 Ok(())
458 }
459}
460
461fn validate_config(config: &Config) -> Result<()> {
462 if config.writers_by_priority.is_empty() {
463 return Err(Error::invalid_config("writers_by_priority cannot be empty"));
464 }
465 let writers = config
466 .writers_by_priority
467 .iter()
468 .copied()
469 .collect::<BTreeSet<_>>();
470 if writers.len() != config.writers_by_priority.len() {
471 return Err(Error::invalid_config("writers_by_priority must be unique"));
472 }
473 if !writers.contains(&WriterId::from_signing_key(&config.signing_key)) {
474 return Err(Error::invalid_config(
475 "writers_by_priority must contain the local writer",
476 ));
477 }
478 Ok(())
479}
480
481fn ensure_authorized(writer: WriterId, writers: &[WriterId]) -> Result<()> {
482 if writers.contains(&writer) {
483 Ok(())
484 } else {
485 Err(Error::invalid_transaction(format!(
486 "writer {writer} is not configured"
487 )))
488 }
489}
490
491fn validate_references(
492 self_id: NodeId,
493 data: &NodeData,
494 nodes: &BTreeSet<NodeId>,
495 objects: &BTreeSet<ObjectId>,
496) -> Result<()> {
497 if let Owner::Node(owner) = data.owner
498 && !nodes.contains(&owner)
499 && owner != self_id
500 {
501 return Err(Error::invalid_input(format!(
502 "owner node {owner} is not locally resolvable"
503 )));
504 }
505 for reference in data.fixed_connections.iter().flatten() {
506 if !nodes.contains(reference) && *reference != self_id {
507 return Err(Error::invalid_input(format!(
508 "fixed connection {reference} is not locally resolvable"
509 )));
510 }
511 }
512 for reference in &data.objects {
513 if !objects.contains(reference) {
514 return Err(Error::invalid_input(format!(
515 "object {reference} is not locally resolvable"
516 )));
517 }
518 }
519 Ok(())
520}