use crate::{
Error, HistoryEntry, Node, NodeData, NodeHistory, NodeId, ObjectId, ObjectPayload, Result,
TransactionId, WriterId,
wire::{self, ParsedTransaction},
};
use chrono::{DateTime, Utc};
use sha2::{Digest, Sha256};
use std::{
collections::BTreeSet,
fs::{self, File, OpenOptions},
io::{Read, Seek, SeekFrom, Write},
path::{Component, Path, PathBuf},
};
mod codec;
use codec::{encode_record, read_record};
const FORMAT: &[u8; 16] = b"KWDBROOT\0\0\0\0\0\0\0\x03";
const STATE_MAGIC: &[u8; 8] = b"KWSTATE3";
const NODE_MAGIC: &[u8; 8] = b"KWNODE03";
const HISTORY_INDEX_MAGIC: &[u8; 8] = b"KWHIDX03";
const HISTORY_ENTRY_MAGIC: &[u8; 8] = b"KWHENT03";
const TX_META_MAGIC: &[u8; 8] = b"KWTMETA3";
const TX_BYTES_MAGIC: &[u8; 8] = b"KWTBYTE3";
const QUEUE_MAGIC: &[u8; 8] = b"KWQUEUE3";
const WAL_MAGIC: &[u8; 8] = b"KWWAL003";
const OBJECT_MAGIC: &[u8; 8] = b"KWOBJ\0\x03\0";
const RECORD_VERSION: u32 = 3;
const OBJECT_HEADER_BYTES: u64 = 8 + 6 + 32 + 8 + 32 + 32;
#[derive(Clone, Debug)]
pub(crate) struct State {
pub format: u32,
pub generation: u64,
pub log_offset: u64,
pub heads: Vec<TransactionId>,
pub writers_by_priority: Vec<WriterId>,
}
#[derive(Clone, Debug)]
pub(crate) struct Candidate {
pub transaction: TransactionId,
pub writer: WriterId,
pub committed_at: DateTime<Utc>,
pub provenance: crate::Provenance,
pub data: NodeData,
}
#[derive(Clone, Debug)]
pub(crate) struct NodeFile {
pub format: u32,
pub id: NodeId,
pub node: Node,
pub visible_transaction: TransactionId,
pub frontier: Vec<Candidate>,
}
#[derive(Clone, Debug)]
pub(crate) struct HistoryIndex {
pub format: u32,
pub node_id: NodeId,
pub frontier: Vec<TransactionId>,
pub visible: Option<TransactionId>,
pub creations: Vec<TransactionId>,
}
#[derive(Clone, Debug)]
pub(crate) struct TxMeta {
pub format: u32,
pub id: TransactionId,
pub parents: Vec<TransactionId>,
pub dag_generation: u64,
}
#[derive(Clone, Copy, Debug)]
enum FileMode {
CreateNew,
Replace,
}
#[derive(Debug)]
struct WalFile {
staged: String,
destination: String,
sha256: [u8; 32],
mode: FileMode,
}
#[derive(Debug)]
struct WalLog {
staged: String,
transaction: TransactionId,
length: u64,
sha256: [u8; 32],
}
#[derive(Debug)]
struct WalIntent {
format: u32,
expected_generation: u64,
expected_log_offset: u64,
log: Option<WalLog>,
files: Vec<WalFile>,
}
#[derive(Debug)]
struct QueueRecord {
format: u32,
transaction: TransactionId,
}
pub(crate) fn open_root(root: &Path, writers_by_priority: &[WriterId]) -> Result<()> {
let format_path = root.join("FORMAT");
if !path_occupied(&format_path) {
let mut nonempty = false;
for entry in read_real_directory(root)? {
let name = entry?.file_name();
if name != "LOCK" {
nonempty = true;
break;
}
}
if nonempty {
return Err(Error::offline_upgrade(
"the nonempty root has no 1.0 FORMAT marker",
));
}
write_sync(&format_path, FORMAT)?;
sync_dir(root)?;
} else if read_regular_bytes(&format_path)? != FORMAT {
return Err(Error::offline_upgrade(
"the root format is not kcode-kweb-db 1.0 root-format 3",
));
}
for directory in [
"wal",
"nodes",
"objects",
"incoming",
"history",
"transactions",
"gossip-outbox",
] {
ensure_real_directory(&root.join(directory))?;
}
clear_incoming(root)?;
let log = root.join("transactions.kwl");
if !path_occupied(&log) {
create_sync(&log, &[])?;
} else {
open_regular_file(&log)?;
}
let state_path = root.join("state.kws");
if !path_occupied(&state_path) {
let state = State {
format: RECORD_VERSION,
generation: 0,
log_offset: 0,
heads: Vec::new(),
writers_by_priority: writers_by_priority.to_vec(),
};
create_sync(&state_path, &encode_record(STATE_MAGIC, &state)?)?;
}
recover_wal(root)?;
recover_outbox_claim(root)?;
let state = read_state(root)?;
if state.format != RECORD_VERSION {
return Err(Error::corrupt("unknown state format version"));
}
if state.writers_by_priority != writers_by_priority {
return Err(Error::invalid_config(
"writers_by_priority does not match the database root",
));
}
Ok(())
}
pub(crate) fn read_state(root: &Path) -> Result<State> {
let state: State = read_record(&root.join("state.kws"), STATE_MAGIC)?;
if state.format != RECORD_VERSION
|| !strictly_sorted_unique(&state.heads)
|| state.writers_by_priority.is_empty()
|| state
.writers_by_priority
.iter()
.copied()
.collect::<BTreeSet<_>>()
.len()
!= state.writers_by_priority.len()
|| state
.writers_by_priority
.iter()
.any(|writer| ed25519_dalek::VerifyingKey::from_bytes(&writer.0).is_err())
{
return Err(Error::corrupt("invalid state record"));
}
Ok(state)
}
pub(crate) fn state_bytes(state: &State) -> Result<Vec<u8>> {
encode_record(STATE_MAGIC, state)
}
pub(crate) fn node_rel(id: NodeId) -> PathBuf {
sharded("nodes", &id.to_string(), "kwn")
}
pub(crate) fn object_rel(id: ObjectId) -> PathBuf {
sharded("objects", &id.to_string(), "kwo")
}
fn incoming_rel(id: TransactionId) -> PathBuf {
PathBuf::from("incoming").join(id.to_string())
}
fn history_dir_rel(id: NodeId) -> PathBuf {
let text = id.to_string();
PathBuf::from("history").join(&text[..2]).join(&text[2..])
}
pub(crate) fn history_index_rel(id: NodeId) -> PathBuf {
history_dir_rel(id).join("index.kwh")
}
pub(crate) fn history_entry_rel(id: NodeId, transaction: TransactionId) -> PathBuf {
history_dir_rel(id)
.join("entries")
.join(format!("{transaction}.khe"))
}
pub(crate) fn tx_meta_rel(id: TransactionId) -> PathBuf {
sharded("transactions", &id.to_string(), "kwi")
}
pub(crate) fn tx_bytes_rel(id: TransactionId) -> PathBuf {
sharded("transactions", &id.to_string(), "kwt")
}
pub(crate) fn outbox_rel(id: TransactionId) -> PathBuf {
PathBuf::from("gossip-outbox").join(format!("{id}.kwq"))
}
fn outbox_claim_rel() -> PathBuf {
PathBuf::from("gossip-outbox").join("INFLIGHT")
}
fn sharded(base: &str, text: &str, extension: &str) -> PathBuf {
PathBuf::from(base)
.join(&text[..2])
.join(format!("{}.{}", &text[2..], extension))
}
pub(crate) fn node_exists(root: &Path, id: NodeId) -> bool {
path_occupied(&root.join(node_rel(id)))
}
pub(crate) fn node_id_occupied(root: &Path, id: NodeId) -> bool {
node_exists(root, id) || path_occupied(&root.join(history_index_rel(id)))
}
pub(crate) fn object_exists(root: &Path, id: ObjectId) -> bool {
path_occupied(&root.join(object_rel(id)))
}
pub(crate) fn tx_exists(root: &Path, id: TransactionId) -> bool {
path_occupied(&root.join(tx_meta_rel(id)))
}
pub(crate) fn read_node(root: &Path, id: NodeId) -> Result<NodeFile> {
let node: NodeFile = read_record(&root.join(node_rel(id)), NODE_MAGIC)?;
let frontier_ids = node
.frontier
.iter()
.map(|candidate| candidate.transaction)
.collect::<Vec<_>>();
let visible = node
.frontier
.iter()
.find(|candidate| candidate.transaction == node.visible_transaction);
if node.format != RECORD_VERSION
|| node.id != id
|| node.node.id != id
|| node.frontier.is_empty()
|| !strictly_sorted_unique(&frontier_ids)
|| visible.is_none()
|| node.node.data.validate().is_err()
|| node.frontier.iter().any(|candidate| {
candidate.provenance.validate().is_err()
|| candidate.data.validate().is_err()
|| ed25519_dalek::VerifyingKey::from_bytes(&candidate.writer.0).is_err()
})
|| visible.is_some_and(|visible| {
node.node.data != visible.data
|| node.node.last_author != visible.provenance.author
|| node.node.committed_at != visible.committed_at
})
{
return Err(Error::corrupt(
"node record identity, version, or contents are invalid",
));
}
Ok(node)
}
pub(crate) fn read_node_optional(root: &Path, id: NodeId) -> Result<Option<NodeFile>> {
match read_node(root, id) {
Ok(node) => Ok(Some(node)),
Err(Error::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(error) => Err(error),
}
}
pub(crate) fn node_bytes(node: &NodeFile) -> Result<Vec<u8>> {
encode_record(NODE_MAGIC, node)
}
pub(crate) fn read_history(root: &Path, id: NodeId) -> Result<NodeHistory> {
let index = read_history_index(root, id)?;
let entries_path = root.join(history_dir_rel(id)).join("entries");
let mut entries = Vec::new();
for entry in read_real_directory(&entries_path)? {
let entry = entry?;
let name = entry.file_name();
let name = name.to_string_lossy();
let transaction: TransactionId = name
.strip_suffix(".khe")
.ok_or_else(|| Error::corrupt("unknown node-history entry"))?
.parse()
.map_err(|_| Error::corrupt("invalid node-history transaction ID"))?;
let history_entry: HistoryEntry = read_record(&entry.path(), HISTORY_ENTRY_MAGIC)?;
if history_entry.transaction_id != transaction || !valid_history_entry(&history_entry) {
return Err(Error::corrupt(
"node-history entry is invalid or does not match its path",
));
}
entries.push(history_entry);
}
Ok(NodeHistory {
node_id: id,
frontier: index.frontier,
visible: index.visible,
entries,
})
}
pub(crate) fn read_history_index(root: &Path, id: NodeId) -> Result<HistoryIndex> {
let history: HistoryIndex =
read_record(&root.join(history_index_rel(id)), HISTORY_INDEX_MAGIC)?;
if history.format != RECORD_VERSION
|| history.node_id != id
|| !strictly_sorted_unique(&history.frontier)
|| !strictly_sorted_unique(&history.creations)
|| history
.visible
.is_some_and(|visible| !history.frontier.contains(&visible))
{
return Err(Error::corrupt(
"history index identity, version, or contents are invalid",
));
}
Ok(history)
}
pub(crate) fn read_history_index_optional(root: &Path, id: NodeId) -> Result<Option<HistoryIndex>> {
match read_history_index(root, id) {
Ok(history) => Ok(Some(history)),
Err(Error::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(error) => Err(error),
}
}
pub(crate) fn read_history_entry_optional(
root: &Path,
id: NodeId,
transaction: TransactionId,
) -> Result<Option<HistoryEntry>> {
match read_record::<HistoryEntry>(
&root.join(history_entry_rel(id, transaction)),
HISTORY_ENTRY_MAGIC,
) {
Ok(entry) => {
if entry.transaction_id != transaction || !valid_history_entry(&entry) {
return Err(Error::corrupt(
"history entry is invalid or does not match its path",
));
}
Ok(Some(entry))
}
Err(Error::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(error) => Err(error),
}
}
pub(crate) fn history_index_bytes(history: &HistoryIndex) -> Result<Vec<u8>> {
encode_record(HISTORY_INDEX_MAGIC, history)
}
pub(crate) fn history_entry_bytes(history: &HistoryEntry) -> Result<Vec<u8>> {
encode_record(HISTORY_ENTRY_MAGIC, history)
}
pub(crate) fn read_tx_meta(root: &Path, id: TransactionId) -> Result<TxMeta> {
let meta: TxMeta = read_record(&root.join(tx_meta_rel(id)), TX_META_MAGIC)?;
if meta.format != RECORD_VERSION || meta.id != id || !strictly_sorted_unique(&meta.parents) {
return Err(Error::corrupt(
"transaction metadata identity, version, or contents are invalid",
));
}
Ok(meta)
}
pub(crate) fn tx_meta_bytes(meta: &TxMeta) -> Result<Vec<u8>> {
encode_record(TX_META_MAGIC, meta)
}
pub(crate) fn read_signed_bytes(root: &Path, id: TransactionId) -> Result<Vec<u8>> {
let path = root.join(tx_bytes_rel(id));
let mut file = open_regular_file(&path)?;
let mut magic = [0; 8];
file.read_exact(&mut magic)?;
if &magic != TX_BYTES_MAGIC {
return Err(Error::corrupt("unknown signed-transaction file version"));
}
let mut stored_id = [0; 32];
file.read_exact(&mut stored_id)?;
if TransactionId(stored_id) != id {
return Err(Error::corrupt("signed-transaction file identity mismatch"));
}
let mut length = [0; 8];
file.read_exact(&mut length)?;
let length = u64::from_be_bytes(length);
if length > crate::model::MAX_TRANSACTION_BYTES as u64 {
return Err(Error::corrupt("stored signed transaction exceeds 64 MiB"));
}
let mut expected_hash = [0; 32];
file.read_exact(&mut expected_hash)?;
let expected_file_length = 8_u64
.checked_add(32 + 8 + 32)
.and_then(|header| header.checked_add(length))
.ok_or_else(|| Error::corrupt("signed-transaction file length overflow"))?;
if file.metadata()?.len() != expected_file_length {
return Err(Error::corrupt(
"signed-transaction file length does not match its header",
));
}
let allocation = usize::try_from(length)
.map_err(|_| Error::corrupt("signed-transaction length does not fit usize"))?;
let mut bytes = Vec::new();
bytes
.try_reserve_exact(allocation)
.map_err(|_| Error::corrupt("cannot allocate stored signed transaction"))?;
bytes.resize(allocation, 0);
file.read_exact(&mut bytes)?;
ensure_eof(&mut file, "signed-transaction file")?;
if Sha256::digest(&bytes)[..] != expected_hash || TransactionId::for_signed_bytes(&bytes) != id
{
return Err(Error::corrupt("signed transaction record mismatch"));
}
Ok(bytes)
}
pub(crate) fn queue_bytes(id: TransactionId) -> Result<Vec<u8>> {
encode_record(
QUEUE_MAGIC,
&QueueRecord {
format: RECORD_VERSION,
transaction: id,
},
)
}
pub(crate) fn transaction_committed(root: &Path, id: TransactionId) -> Result<bool> {
if !tx_exists(root, id) {
return Ok(false);
}
read_tx_meta(root, id)?;
Ok(true)
}
pub(crate) fn dag_generation(root: &Path, id: TransactionId) -> Result<u64> {
Ok(read_tx_meta(root, id)?.dag_generation)
}
pub(crate) fn is_ancestor(
root: &Path,
ancestor: TransactionId,
descendant: TransactionId,
overlay: Option<&DagOverlay<'_>>,
) -> Result<bool> {
if ancestor == descendant {
return Ok(false);
}
let mut stack = parents_for(root, descendant, overlay)?;
let mut visited = BTreeSet::new();
while let Some(current) = stack.pop() {
if current == ancestor {
return Ok(true);
}
if visited.insert(current) {
stack.extend(parents_for(root, current, overlay)?);
}
}
Ok(false)
}
pub(crate) struct DagOverlay<'a> {
pub id: TransactionId,
pub parents: &'a [TransactionId],
}
fn parents_for(
root: &Path,
id: TransactionId,
overlay: Option<&DagOverlay<'_>>,
) -> Result<Vec<TransactionId>> {
if let Some(overlay) = overlay
&& id == overlay.id
{
return Ok(overlay.parents.to_vec());
}
Ok(read_tx_meta(root, id)?.parents)
}
pub(crate) fn read_object(root: &Path, id: ObjectId) -> Result<(TransactionId, Vec<u8>)> {
let mut file = open_regular_file(&root.join(object_rel(id)))?;
let (creator, length, expected_hash) = read_object_header(&mut file, id)?;
let expected_file_length = OBJECT_HEADER_BYTES
.checked_add(length)
.ok_or_else(|| Error::corrupt("object file length overflow"))?;
if file.metadata()?.len() != expected_file_length {
return Err(Error::corrupt(
"object file length does not match its header",
));
}
let allocation = usize::try_from(length)
.map_err(|_| Error::corrupt("object length does not fit usize on this target"))?;
let mut bytes = Vec::new();
bytes
.try_reserve_exact(allocation)
.map_err(|_| Error::Io(std::io::Error::other("object allocation failed")))?;
bytes.resize(allocation, 0);
file.read_exact(&mut bytes)?;
ensure_eof(&mut file, "object file")?;
if Sha256::digest(&bytes)[..] != expected_hash {
return Err(Error::corrupt("object payload checksum mismatch"));
}
Ok((creator, bytes))
}
pub(crate) fn object_creator(root: &Path, id: ObjectId) -> Result<TransactionId> {
let mut file = open_regular_file(&root.join(object_rel(id)))?;
let (creator, length, _) = read_object_header(&mut file, id)?;
let expected_file_length = OBJECT_HEADER_BYTES
.checked_add(length)
.ok_or_else(|| Error::corrupt("object file length overflow"))?;
if file.metadata()?.len() != expected_file_length {
return Err(Error::corrupt(
"object file length does not match its header",
));
}
Ok(creator)
}
fn read_object_header(file: &mut File, id: ObjectId) -> Result<(TransactionId, u64, [u8; 32])> {
let mut computed_hash = Sha256::new();
let mut magic = [0; 8];
file.read_exact(&mut magic)?;
if &magic != OBJECT_MAGIC {
return Err(Error::corrupt("unknown object envelope version"));
}
computed_hash.update(magic);
let mut stored_id = [0; 6];
file.read_exact(&mut stored_id)?;
if ObjectId(stored_id) != id || !id.valid_domain() {
return Err(Error::corrupt("object envelope identity mismatch"));
}
computed_hash.update(stored_id);
let mut creator = [0; 32];
file.read_exact(&mut creator)?;
computed_hash.update(creator);
let mut length = [0; 8];
file.read_exact(&mut length)?;
computed_hash.update(length);
let length = u64::from_be_bytes(length);
if length > crate::MAX_OBJECT_BYTES {
return Err(Error::corrupt("object envelope exceeds 32 GiB"));
}
let mut hash = [0; 32];
file.read_exact(&mut hash)?;
computed_hash.update(hash);
let mut stored_header_hash = [0; 32];
file.read_exact(&mut stored_header_hash)?;
if computed_hash.finalize()[..] != stored_header_hash {
return Err(Error::corrupt("object envelope header checksum mismatch"));
}
Ok((TransactionId(creator), length, hash))
}
pub(crate) fn log_frame_length(transaction_length: usize) -> Result<u64> {
let length = u64::try_from(transaction_length)
.map_err(|_| Error::invalid_input("transaction length does not fit u64"))?;
length
.checked_add(8 + 32 + 32)
.ok_or_else(|| Error::invalid_input("transaction log frame length overflow"))
}
pub(crate) struct SpooledObject {
pub id: ObjectId,
path: PathBuf,
envelope_sha256: [u8; 32],
}
pub(crate) fn spool_objects(
root: &Path,
creator: TransactionId,
objects: Vec<ObjectPayload>,
) -> Result<Vec<SpooledObject>> {
let relative = incoming_rel(creator);
let directory = root.join(&relative);
fs::create_dir(&directory)?;
let result = (|| {
let mut spooled = Vec::with_capacity(objects.len());
for payload in objects {
let path = directory.join(format!("{}.kwo", payload.id));
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(&path)?;
let envelope_sha256 = write_object_envelope(&mut file, creator, &payload)?;
file.sync_all()?;
spooled.push(SpooledObject {
id: payload.id,
path,
envelope_sha256,
});
}
sync_dir(&directory)?;
sync_dir(&root.join("incoming"))?;
Ok(spooled)
})();
if result.is_err() {
let _ = fs::remove_dir_all(&directory);
let _ = sync_dir(&root.join("incoming"));
}
result
}
pub(crate) fn discard_spooled_objects(root: &Path, creator: TransactionId) -> Result<()> {
let directory = root.join(incoming_rel(creator));
match fs::remove_dir_all(&directory) {
Ok(()) => sync_dir(&root.join("incoming")),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(error.into()),
}
}
pub(crate) struct Commit {
root: PathBuf,
building: PathBuf,
prepared: PathBuf,
intent: WalIntent,
prepared_published: bool,
}
pub(crate) struct CommitFailure {
pub error: Error,
pub prepared: bool,
}
impl Commit {
pub(crate) fn new(root: &Path, label: TransactionId, state: &State) -> Result<Self> {
let generation = state
.generation
.checked_add(1)
.ok_or_else(|| Error::corrupt("database generation overflow"))?;
let suffix = format!("{generation:020}-{label}");
let building = root.join("wal").join(format!(".building-{suffix}"));
let prepared = root.join("wal").join(format!("{suffix}.prepared"));
fs::create_dir(&building)?;
Ok(Self {
root: root.to_path_buf(),
building,
prepared,
intent: WalIntent {
format: RECORD_VERSION,
expected_generation: state.generation,
expected_log_offset: state.log_offset,
log: None,
files: Vec::new(),
},
prepared_published: false,
})
}
pub(crate) fn stage_bytes(
&mut self,
destination: PathBuf,
bytes: &[u8],
create_new: bool,
) -> Result<()> {
let name = format!("{:08}.stage", self.intent.files.len());
let staged = self.building.join(&name);
create_sync(&staged, bytes)?;
self.intent.files.push(WalFile {
staged: name,
destination: relative_text(&destination)?,
sha256: Sha256::digest(bytes).into(),
mode: if create_new {
FileMode::CreateNew
} else {
FileMode::Replace
},
});
Ok(())
}
pub(crate) fn stage_object(
&mut self,
destination: PathBuf,
creator: TransactionId,
payload: &ObjectPayload,
) -> Result<()> {
let name = format!("{:08}.stage", self.intent.files.len());
let staged = self.building.join(&name);
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(&staged)?;
let file_hash = write_object_envelope(&mut file, creator, payload)?;
file.sync_all()?;
self.intent.files.push(WalFile {
staged: name,
destination: relative_text(&destination)?,
sha256: file_hash,
mode: FileMode::CreateNew,
});
Ok(())
}
pub(crate) fn stage_spooled_object(
&mut self,
destination: PathBuf,
object: &SpooledObject,
) -> Result<()> {
let name = format!("{:08}.stage", self.intent.files.len());
let staged = self.building.join(&name);
fs::hard_link(&object.path, &staged)?;
if hash_file(&staged)? != object.envelope_sha256 {
return Err(Error::corrupt("spooled object checksum mismatch"));
}
self.intent.files.push(WalFile {
staged: name,
destination: relative_text(&destination)?,
sha256: object.envelope_sha256,
mode: FileMode::CreateNew,
});
Ok(())
}
pub(crate) fn stage_signed_transaction(
&mut self,
destination: PathBuf,
id: TransactionId,
transaction: &[u8],
) -> Result<()> {
if transaction.len() > crate::model::MAX_TRANSACTION_BYTES {
return Err(Error::invalid_input("signed transaction exceeds 64 MiB"));
}
let name = format!("{:08}.stage", self.intent.files.len());
let staged = self.building.join(&name);
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(&staged)?;
let mut file_hash = Sha256::new();
let transaction_hash: [u8; 32] = Sha256::digest(transaction).into();
write_hashed(&mut file, &mut file_hash, TX_BYTES_MAGIC)?;
write_hashed(&mut file, &mut file_hash, &id.0)?;
write_hashed(
&mut file,
&mut file_hash,
&(transaction.len() as u64).to_be_bytes(),
)?;
write_hashed(&mut file, &mut file_hash, &transaction_hash)?;
write_hashed(&mut file, &mut file_hash, transaction)?;
file.sync_all()?;
self.intent.files.push(WalFile {
staged: name,
destination: relative_text(&destination)?,
sha256: file_hash.finalize().into(),
mode: FileMode::CreateNew,
});
Ok(())
}
pub(crate) fn stage_log(&mut self, id: TransactionId, transaction: &[u8]) -> Result<()> {
if self.intent.log.is_some() {
return Err(Error::corrupt("WAL already contains a log frame"));
}
let name = "LOG.stage".to_owned();
let staged = self.building.join(&name);
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(&staged)?;
let mut frame_hash = Sha256::new();
write_hashed(
&mut file,
&mut frame_hash,
&(transaction.len() as u64).to_be_bytes(),
)?;
write_hashed(&mut file, &mut frame_hash, &id.0)?;
write_hashed(&mut file, &mut frame_hash, &Sha256::digest(transaction))?;
write_hashed(&mut file, &mut frame_hash, transaction)?;
file.sync_all()?;
self.intent.log = Some(WalLog {
staged: name,
transaction: id,
length: log_frame_length(transaction.len())?,
sha256: frame_hash.finalize().into(),
});
Ok(())
}
pub(crate) fn finish(mut self) -> std::result::Result<(), CommitFailure> {
if self.intent.format != RECORD_VERSION {
return Err(CommitFailure {
error: Error::corrupt("invalid WAL intent version"),
prepared: false,
});
}
let intent = encode_record(WAL_MAGIC, &self.intent).map_err(|error| CommitFailure {
error,
prepared: false,
})?;
create_sync(&self.building.join("INTENT"), &intent).map_err(|error| CommitFailure {
error,
prepared: false,
})?;
sync_dir(&self.building).map_err(|error| CommitFailure {
error,
prepared: false,
})?;
fs::rename(&self.building, &self.prepared).map_err(|error| CommitFailure {
error: error.into(),
prepared: false,
})?;
self.prepared_published = true;
sync_dir(&self.root.join("wal")).map_err(|error| CommitFailure {
error,
prepared: true,
})?;
if let Err(first_error) = apply_prepared(&self.root, &self.prepared)
&& let Err(second_error) = apply_prepared(&self.root, &self.prepared)
{
return Err(CommitFailure {
error: Error::corrupt(format!(
"prepared WAL application failed twice: {first_error}; {second_error}"
)),
prepared: true,
});
}
if fs::remove_dir_all(&self.prepared).is_ok() {
let _ = sync_dir(&self.root.join("wal"));
}
Ok(())
}
}
impl Drop for Commit {
fn drop(&mut self) {
if !self.prepared_published {
let _ = fs::remove_dir_all(&self.building);
}
}
}
fn recover_wal(root: &Path) -> Result<()> {
let wal = root.join("wal");
let mut prepared = Vec::new();
for entry in read_real_directory(&wal)? {
let entry = entry?;
let name = entry.file_name();
let name = name.to_string_lossy();
ensure_real_directory(&entry.path())?;
if name.starts_with(".building-") {
fs::remove_dir_all(entry.path())?;
} else if name.ends_with(".prepared") {
prepared.push(entry.path());
} else {
return Err(Error::corrupt("unknown entry in WAL directory"));
}
}
prepared.sort();
for path in prepared {
apply_prepared(root, &path)?;
fs::remove_dir_all(path)?;
sync_dir(&wal)?;
}
Ok(())
}
fn apply_prepared(root: &Path, prepared: &Path) -> Result<()> {
let intent: WalIntent = read_record(&prepared.join("INTENT"), WAL_MAGIC)?;
if intent.format != RECORD_VERSION {
return Err(Error::corrupt("unknown WAL format version"));
}
let state = read_state(root)?;
let committed_generation = intent
.expected_generation
.checked_add(1)
.ok_or_else(|| Error::corrupt("WAL generation overflow"))?;
if state.generation > committed_generation {
return Ok(());
}
if state.generation != intent.expected_generation && state.generation != committed_generation {
return Err(Error::corrupt("WAL expected generation mismatch"));
}
let committed_log_offset = match &intent.log {
Some(log) => intent
.expected_log_offset
.checked_add(log.length)
.ok_or_else(|| Error::corrupt("WAL log offset overflow"))?,
None => intent.expected_log_offset,
};
let expected_state_log_offset = if state.generation == intent.expected_generation {
intent.expected_log_offset
} else {
committed_log_offset
};
if state.log_offset != expected_state_log_offset {
return Err(Error::corrupt("WAL state log offset mismatch"));
}
if let Some(log) = &intent.log {
apply_log(root, prepared, intent.expected_log_offset, log)?;
}
let mut state_operation = None;
for operation in &intent.files {
if operation.destination == "state.kws" {
if state_operation.replace(operation).is_some() {
return Err(Error::corrupt("WAL contains multiple state operations"));
}
} else {
apply_file_operation(root, prepared, operation)?;
}
}
let state_operation =
state_operation.ok_or_else(|| Error::corrupt("WAL contains no state operation"))?;
apply_file_operation(root, prepared, state_operation)?;
Ok(())
}
fn apply_file_operation(root: &Path, prepared: &Path, operation: &WalFile) -> Result<()> {
let destination = checked_join(root, &operation.destination)?;
let staged = checked_join(prepared, &operation.staged)?;
ensure_real_parent_directories(root, &destination)?;
if !path_occupied(&staged) {
if !path_occupied(&destination) || hash_file(&destination)? != operation.sha256 {
return Err(Error::corrupt("applied WAL file does not match its intent"));
}
return Ok(());
}
if hash_file(&staged)? != operation.sha256 {
return Err(Error::corrupt("staged WAL file checksum mismatch"));
}
match operation.mode {
FileMode::CreateNew => {
if path_occupied(&destination) {
if hash_file(&destination)? != operation.sha256 {
return Err(Error::corrupt("create-new WAL destination collision"));
}
fs::remove_file(&staged)?;
} else {
fs::hard_link(&staged, &destination)?;
fs::remove_file(&staged)?;
}
}
FileMode::Replace => fs::rename(&staged, &destination)?,
}
sync_parent(&destination)
}
fn apply_log(root: &Path, prepared: &Path, offset: u64, log: &WalLog) -> Result<()> {
let path = root.join("transactions.kwl");
let mut file = open_regular_options(&path, true)?;
let length = file.metadata()?.len();
let staged_path = checked_join(prepared, &log.staged)?;
let mut staged = open_regular_file(&staged_path)?;
if staged.metadata()?.len() != log.length || hash_file(&staged_path)? != log.sha256 {
return Err(Error::corrupt("staged transaction log frame mismatch"));
}
let mut declared_length = [0; 8];
let mut declared_transaction = [0; 32];
staged.read_exact(&mut declared_length)?;
staged.read_exact(&mut declared_transaction)?;
let declared_length = u64::from_be_bytes(declared_length);
if declared_length.checked_add(8 + 32 + 32) != Some(log.length)
|| TransactionId(declared_transaction) != log.transaction
{
return Err(Error::corrupt(
"staged transaction log frame header mismatch",
));
}
staged.seek(SeekFrom::Start(0))?;
if length == offset {
file.seek(SeekFrom::Start(offset))?;
std::io::copy(&mut staged, &mut file)?;
file.sync_all()?;
} else if length == offset + log.length {
file.seek(SeekFrom::Start(offset))?;
if hash_reader(std::io::Read::by_ref(&mut file).take(log.length))? != log.sha256 {
return Err(Error::corrupt(
"transaction log frame differs at WAL offset",
));
}
} else {
return Err(Error::corrupt(
"unexpected transaction log layout during WAL recovery",
));
}
Ok(())
}
pub(crate) fn next_outbox(root: &Path) -> Result<Option<TransactionId>> {
next_queue_entry(&root.join("gossip-outbox"))
}
pub(crate) fn claim_outbox(root: &Path, id: TransactionId) -> Result<()> {
let source = root.join(outbox_rel(id));
let destination = root.join(outbox_claim_rel());
if path_occupied(&destination) {
return Err(Error::corrupt(
"gossip outbox already has an in-flight item",
));
}
fs::rename(source, destination)?;
sync_dir(&root.join("gossip-outbox"))
}
pub(crate) fn requeue_outbox_claim(root: &Path, id: TransactionId) -> Result<()> {
let source = root.join(outbox_claim_rel());
let destination = root.join(outbox_rel(id));
if path_occupied(&destination) {
return Err(Error::corrupt(
"gossip outbox destination is already occupied",
));
}
fs::rename(source, destination)?;
sync_dir(&root.join("gossip-outbox"))
}
pub(crate) fn acknowledge_outbox_claim(root: &Path) -> Result<()> {
let path = root.join(outbox_claim_rel());
fs::remove_file(&path)?;
sync_parent(&path)
}
fn next_queue_entry(directory: &Path) -> Result<Option<TransactionId>> {
for entry in read_real_directory(directory)? {
let entry = entry?;
let name = entry.file_name();
let name = name.to_string_lossy();
if name == "INFLIGHT" {
continue;
}
let id: TransactionId = name
.strip_suffix(".kwq")
.ok_or_else(|| Error::corrupt("unknown durable queue entry"))?
.parse()
.map_err(|_| Error::corrupt("invalid durable queue transaction ID"))?;
read_queue_record(&entry.path(), id)?;
return Ok(Some(id));
}
Ok(None)
}
fn read_queue_record(path: &Path, expected: TransactionId) -> Result<()> {
let record: QueueRecord = read_record(path, QUEUE_MAGIC)?;
if record.format != RECORD_VERSION || record.transaction != expected {
return Err(Error::corrupt(
"durable queue record identity or version mismatch",
));
}
Ok(())
}
pub(crate) fn recover_outbox_claim(root: &Path) -> Result<()> {
let claim = root.join(outbox_claim_rel());
let record: QueueRecord = match read_record(&claim, QUEUE_MAGIC) {
Ok(record) => record,
Err(Error::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(error) => return Err(error),
};
if record.format != RECORD_VERSION {
return Err(Error::corrupt("unknown in-flight outbox record version"));
}
let destination = root.join(outbox_rel(record.transaction));
if path_occupied(&destination) {
return Err(Error::corrupt(
"in-flight and queued outbox records both exist",
));
}
fs::rename(claim, destination)?;
sync_dir(&root.join("gossip-outbox"))
}
pub(crate) fn package_for(root: &Path, id: TransactionId) -> Result<crate::TransactionPackage> {
let transaction = read_signed_bytes(root, id)?;
let parsed = wire::parse_signed(&transaction)?;
let mut objects = Vec::with_capacity(parsed.unsigned.objects.len());
for declaration in &parsed.unsigned.objects {
let (creator, bytes) = read_object(root, declaration.id)?;
if creator != id
|| bytes.len() as u64 != declaration.length
|| wire::object_hash(&bytes) != declaration.sha256
{
return Err(Error::corrupt(
"outbox object does not match transaction declaration",
));
}
objects.push(ObjectPayload {
id: declaration.id,
bytes,
});
}
Ok(crate::TransactionPackage {
transaction,
objects,
})
}
pub(crate) fn history_entry(
parsed: &ParsedTransaction,
data: NodeData,
created: bool,
) -> HistoryEntry {
HistoryEntry {
transaction_id: parsed.id,
writer: parsed.unsigned.writer,
committed_at: parsed.unsigned.committed_at,
provenance: parsed.unsigned.provenance.clone(),
active: true,
created,
updated: !created,
data: Some(data),
merge_pairs: parsed.unsigned.merge_pairs.clone(),
}
}
pub(crate) fn empty_history(id: NodeId) -> HistoryIndex {
HistoryIndex {
format: RECORD_VERSION,
node_id: id,
frontier: Vec::new(),
visible: None,
creations: Vec::new(),
}
}
pub(crate) fn record_version() -> u32 {
RECORD_VERSION
}
fn valid_history_entry(entry: &HistoryEntry) -> bool {
entry.active
&& entry.provenance.validate().is_ok()
&& entry.created != entry.updated
&& entry
.data
.as_ref()
.is_some_and(|data| data.validate().is_ok())
&& strictly_sorted_unique(&entry.merge_pairs)
&& entry
.merge_pairs
.iter()
.all(|pair| pair.first < pair.second)
&& ed25519_dalek::VerifyingKey::from_bytes(&entry.writer.0).is_ok()
}
fn strictly_sorted_unique<T: Ord>(values: &[T]) -> bool {
values.windows(2).all(|pair| pair[0] < pair[1])
}
fn write_sync(path: &Path, bytes: &[u8]) -> Result<()> {
let mut file = OpenOptions::new().write(true).create_new(true).open(path)?;
file.write_all(bytes)?;
file.sync_all()?;
Ok(())
}
fn create_sync(path: &Path, bytes: &[u8]) -> Result<()> {
write_sync(path, bytes)
}
fn hash_file(path: &Path) -> Result<[u8; 32]> {
hash_reader(open_regular_file(path)?)
}
fn hash_reader(mut reader: impl Read) -> Result<[u8; 32]> {
let mut hash = Sha256::new();
let mut buffer = [0_u8; 64 * 1024];
loop {
let read = reader.read(&mut buffer)?;
if read == 0 {
break;
}
hash.update(&buffer[..read]);
}
Ok(hash.finalize().into())
}
fn write_hashed(file: &mut File, hash: &mut Sha256, bytes: &[u8]) -> Result<()> {
file.write_all(bytes)?;
hash.update(bytes);
Ok(())
}
fn write_object_envelope(
file: &mut File,
creator: TransactionId,
payload: &ObjectPayload,
) -> Result<[u8; 32]> {
let mut file_hash = Sha256::new();
let payload_hash = wire::object_hash(&payload.bytes);
let mut header = Vec::with_capacity((OBJECT_HEADER_BYTES - 32) as usize);
header.extend_from_slice(OBJECT_MAGIC);
header.extend_from_slice(&payload.id.0);
header.extend_from_slice(&creator.0);
header.extend_from_slice(&(payload.bytes.len() as u64).to_be_bytes());
header.extend_from_slice(&payload_hash);
let header_hash = Sha256::digest(&header);
write_hashed(file, &mut file_hash, &header)?;
write_hashed(file, &mut file_hash, &header_hash)?;
write_hashed(file, &mut file_hash, &payload.bytes)?;
Ok(file_hash.finalize().into())
}
pub(crate) fn clear_incoming(root: &Path) -> Result<()> {
let directory = root.join("incoming");
for entry in read_real_directory(&directory)? {
let entry = entry?;
let metadata = fs::symlink_metadata(entry.path())?;
if metadata.file_type().is_symlink() || !metadata.file_type().is_dir() {
return Err(Error::corrupt("unknown entry in incoming object spool"));
}
fs::remove_dir_all(entry.path())?;
}
sync_dir(&directory)
}
fn ensure_eof(file: &mut File, kind: &str) -> Result<()> {
let mut trailing = [0_u8; 1];
if file.read(&mut trailing)? == 0 {
Ok(())
} else {
Err(Error::corrupt(format!("{kind} grew while it was read")))
}
}
fn relative_text(path: &Path) -> Result<String> {
if path.is_absolute()
|| path
.components()
.any(|component| !matches!(component, Component::Normal(_)))
{
return Err(Error::corrupt(
"WAL destination is not a safe relative path",
));
}
path.to_str()
.map(str::to_owned)
.ok_or_else(|| Error::corrupt("WAL destination is not UTF-8"))
}
fn checked_join(root: &Path, relative: &str) -> Result<PathBuf> {
let path = Path::new(relative);
relative_text(path)?;
Ok(root.join(path))
}
pub(crate) fn path_occupied(path: &Path) -> bool {
match fs::symlink_metadata(path) {
Ok(_) => true,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => false,
Err(_) => true,
}
}
fn open_regular_file(path: &Path) -> Result<File> {
open_regular_options(path, false)
}
fn open_regular_options(path: &Path, write: bool) -> Result<File> {
let metadata = fs::symlink_metadata(path)?;
if metadata.file_type().is_symlink() || !metadata.file_type().is_file() {
return Err(Error::corrupt(format!(
"expected a regular file at {}",
path.display()
)));
}
let file = OpenOptions::new().read(true).write(write).open(path)?;
if !file.metadata()?.file_type().is_file() {
return Err(Error::corrupt(format!(
"opened path is not a regular file: {}",
path.display()
)));
}
Ok(file)
}
fn read_regular_bytes(path: &Path) -> Result<Vec<u8>> {
let mut file = open_regular_file(path)?;
let length = usize::try_from(file.metadata()?.len())
.map_err(|_| Error::corrupt("file length does not fit usize"))?;
let mut bytes = Vec::new();
bytes
.try_reserve_exact(length)
.map_err(|_| Error::Io(std::io::Error::other("file allocation failed")))?;
bytes.resize(length, 0);
file.read_exact(&mut bytes)?;
let mut trailing = [0_u8; 1];
if file.read(&mut trailing)? != 0 {
return Err(Error::corrupt("file grew while it was read"));
}
Ok(bytes)
}
fn read_real_directory(path: &Path) -> Result<fs::ReadDir> {
let metadata = fs::symlink_metadata(path)?;
if metadata.file_type().is_symlink() || !metadata.file_type().is_dir() {
return Err(Error::corrupt(format!(
"expected a real directory at {}",
path.display()
)));
}
Ok(fs::read_dir(path)?)
}
fn ensure_real_directory(path: &Path) -> Result<()> {
match fs::symlink_metadata(path) {
Ok(metadata) => {
if metadata.file_type().is_symlink() || !metadata.file_type().is_dir() {
return Err(Error::corrupt(format!(
"expected a real directory at {}",
path.display()
)));
}
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
let parent = path
.parent()
.ok_or_else(|| Error::corrupt("directory path has no parent"))?;
let parent_metadata = fs::symlink_metadata(parent)?;
if parent_metadata.file_type().is_symlink() || !parent_metadata.file_type().is_dir() {
return Err(Error::corrupt(format!(
"directory parent is not real: {}",
parent.display()
)));
}
fs::create_dir(path)?;
File::open(parent)?.sync_all()?;
}
Err(error) => return Err(error.into()),
}
Ok(())
}
fn ensure_real_parent_directories(root: &Path, destination: &Path) -> Result<()> {
let parent = destination
.parent()
.ok_or_else(|| Error::corrupt("destination has no parent"))?;
let relative = parent
.strip_prefix(root)
.map_err(|_| Error::corrupt("WAL destination escaped the database root"))?;
let mut current = root.to_path_buf();
ensure_real_directory(¤t)?;
for component in relative.components() {
let Component::Normal(component) = component else {
return Err(Error::corrupt("WAL destination parent is not canonical"));
};
current.push(component);
ensure_real_directory(¤t)?;
}
Ok(())
}
fn sync_parent(path: &Path) -> Result<()> {
sync_dir(
path.parent()
.ok_or_else(|| Error::corrupt("path has no parent"))?,
)
}
fn sync_dir(path: &Path) -> Result<()> {
ensure_real_directory(path)?;
File::open(path)?.sync_all()?;
Ok(())
}