use futures::{future, stream, FutureExt, StreamExt};
use litl::{impl_debug_as_litl, impl_nested_tagged_data_serde, NestedTaggedData};
use ridl::{
hashing::{HashOf, StrongHash, StrongHasher},
signing::{SignatureError, Signed, SignerID, SignerSecret},
};
use serde::Serialize;
use serde_derive::{Deserialize, Serialize};
use thiserror::Error;
use tracing::{debug, error, info};
use crate::{
telepathic::{
ApplyDiffErrorFor, ApplyDiffResult, ApplyDiffSuccess, Telepathic, TelepathicDiff,
},
ObjectID, StorageBackend,
};
#[derive(Clone, Serialize, Deserialize)]
pub struct LogHeader {
pub appender: SignerID,
pub meta: Option<litl::Val>,
}
impl_debug_as_litl!(LogHeader);
#[derive(Debug)]
pub struct LogState {
pub id: LogID,
pub header: Option<LogHeader>,
pub entries: Vec<litl::Val>,
pub last_hash: Option<Signed<StrongHash>>,
hasher: StrongHasher,
}
impl LogState {
pub(crate) fn new_empty(id: LogID) -> Self {
Self {
id,
header: None,
entries: Vec::new(),
last_hash: None,
hasher: StrongHasher::default(),
}
}
pub(crate) fn new<M: Serialize>(meta: Option<M>) -> (Self, LogWriteAccess) {
let signer_secret = SignerSecret::new_random();
let header = LogHeader {
appender: signer_secret.pub_id(),
meta: meta.map(|meta| litl::to_val(&meta).unwrap()),
};
let id = LogID(HashOf::hash(&header));
(
Self {
id,
header: Some(header),
entries: Vec::new(),
last_hash: None,
hasher: StrongHasher::default(),
},
LogWriteAccess(signer_secret),
)
}
pub(crate) fn diff_for_new_entry(
&self,
entry: litl::Val,
write_access: &LogWriteAccess,
) -> LogDiff {
let mut hasher = self.hasher.clone();
hasher.update(&litl::to_vec_canonical(&entry).unwrap());
LogDiff {
id: self.id,
header: None,
after: self.entries.len(),
new_entries: vec![entry],
new_hash: Some(write_access.0.sign(hasher.finalize())),
}
}
}
#[derive(Copy, Clone, PartialEq, Eq, Hash)]
pub struct LogID(pub HashOf<LogHeader>);
impl NestedTaggedData for LogID {
const TAG: &'static str = "log";
type Inner = HashOf<LogHeader>;
fn as_inner(&self) -> &Self::Inner {
&self.0
}
fn from_inner(inner: Self::Inner) -> Self
where
Self: Sized,
{
LogID(inner)
}
}
impl_nested_tagged_data_serde!(LogID);
impl_debug_as_litl!(LogID);
pub struct LogWriteAccess(pub(crate) SignerSecret);
impl NestedTaggedData for LogWriteAccess {
const TAG: &'static str = "logWriteAccess";
type Inner = SignerSecret;
fn as_inner(&self) -> &Self::Inner {
&self.0
}
fn from_inner(inner: Self::Inner) -> Self
where
Self: Sized,
{
LogWriteAccess(inner)
}
}
impl_nested_tagged_data_serde!(LogWriteAccess);
impl_debug_as_litl!(LogWriteAccess);
#[derive(Clone, Serialize, Deserialize, Debug)]
pub struct LogDiff {
pub id: LogID,
pub header: Option<LogHeader>,
pub after: usize,
pub new_entries: Vec<litl::Val>,
pub new_hash: Option<Signed<StrongHash>>,
}
impl TelepathicDiff for LogDiff {
type ID = LogID;
fn id(&self) -> Self::ID {
self.id
}
}
pub type LogStateInfo = usize;
#[derive(Serialize, Deserialize)]
pub struct StorageHashAndLen {
pub hash: Signed<StrongHash>,
pub len: usize,
}
impl Telepathic for LogState {
type ID = LogID;
type WriteAccess = LogWriteAccess;
type StateInfo = LogStateInfo;
type Diff = LogDiff;
type Error = LogError;
fn id(&self) -> Self::ID {
self.id
}
fn try_apply_diff(
&mut self,
diff: LogDiff,
) -> ApplyDiffResult<Self::StateInfo, Self::ID, Self::Diff, Self::Error> {
debug_assert_eq!(diff.id, self.id);
let (header, got_header_first_time) = match (&mut self.header, &diff.header) {
(None, None) => {
return Ok(None)
}
(own @ None, Some(diff_header)) => {
if HashOf::hash(diff_header) != self.id.0 {
return Err(LogError::InvalidHeaderHash.into());
}
*own = Some(diff_header.clone());
(own.as_ref().unwrap(), true)
}
(Some(own_header), _) => (&*own_header, false),
};
if diff.after == 0 && diff.new_entries.is_empty() {
if self.entries.is_empty() {
return Ok(Some(ApplyDiffSuccess {
new_state_info: 0,
effective_diff: LogDiff {
id: diff.id,
header: if got_header_first_time {
self.header.clone()
} else {
None
},
after: 0,
new_entries: vec![],
new_hash: None,
},
}));
} else {
return Err(ApplyDiffErrorFor::InvalidKnownStateAssumption(
ObjectID::Log(self.id),
"Got initial diff for non-empty log".to_owned(),
));
}
}
let new_hash = diff.new_hash.as_ref().expect("Expected new hash on append");
new_hash
.ensure_signed_by(&header.appender)
.map_err(LogError::InvalidSignature)?;
if diff.after > self.entries.len() {
return Err(ApplyDiffErrorFor::InvalidKnownStateAssumption(
ObjectID::Log(self.id),
"Got diff later than current log length".to_owned(),
));
}
if diff.after + diff.new_entries.len() <= self.entries.len() {
info!(
after = diff.after,
new_entries = diff.new_entries.len(),
self_len = self.entries.len(),
"Received completely redundant log update"
);
return Ok(None)
}
let overlap = self.entries.len() - diff.after;
let data_len_before = self.entries.len();
let effective_new_entries = &diff.new_entries[overlap..];
let mut new_hasher = self.hasher.clone();
for entry in &effective_new_entries[..] {
new_hasher.update(&litl::to_vec_canonical(&entry).unwrap());
}
if new_hasher.finalize() == new_hash.attested {
self.entries.extend(effective_new_entries.to_vec());
self.last_hash = Some(new_hash.clone());
self.hasher = new_hasher;
let effective_diff = if overlap > 0 {
info!(
overlap = overlap,
after = diff.after,
self_len = data_len_before,
new_entries = diff.new_entries.len(),
effective_new_entries = effective_new_entries.len(),
"Received redundant log update"
);
LogDiff {
id: diff.id,
header: if got_header_first_time {
self.header.clone()
} else {
None
},
after: data_len_before,
new_entries: effective_new_entries.to_vec(),
new_hash: diff.new_hash,
}
} else {
LogDiff {
header: if got_header_first_time {
self.header.clone()
} else {
None
},
..diff
}
};
Ok(Some(ApplyDiffSuccess {
new_state_info: self.entries.len(),
effective_diff,
}))
} else {
Err(ApplyDiffErrorFor::Other(LogError::InvalidHash))
}
}
fn diff_since(&self, state_info: Option<&Self::StateInfo>) -> Option<Self::Diff> {
let (has_header, state_len) = match state_info {
None => (false, 0),
Some(len) => (true, *len as usize),
};
if state_len >= self.entries.len() && has_header {
None
} else {
Some(LogDiff {
id: self.id,
header: if has_header {
None
} else {
self.header.clone()
},
after: state_len,
new_entries: self.entries[state_len..].to_vec(),
new_hash: self.last_hash.clone(),
})
}
}
fn state_info(&self) -> Option<Self::StateInfo> {
if self.header.is_none() {
None
} else {
Some(self.entries.len())
}
}
fn load(
id: LogID,
storage: Box<dyn StorageBackend>,
) -> std::pin::Pin<Box<dyn futures::Stream<Item = Self::Diff>>> {
let key = id.to_string();
let header_stream = {
let key = key.clone();
stream::once(storage.get_key(&format!("header_{}", key))).filter_map(
move |maybe_bytes| {
future::ready(maybe_bytes.and_then(|bytes| {
litl::from_slice(&bytes)
.map_err(|err| {
error!(err = ?err, id = ?id, "Failed to read loaded log header");
err
})
.ok()
.map(|header| LogDiff {
id,
header: Some(header),
after: 0,
new_entries: vec![],
new_hash: None,
})
}))
},
)
};
let all_chunks_stream = {
stream::once(storage.get_key(&format!("hash_and_len_{}", key)))
.filter_map(move |maybe_hash_and_len_bytes| {
future::ready(maybe_hash_and_len_bytes.and_then(|hash_and_len_bytes| {
litl::from_slice::<StorageHashAndLen>(&hash_and_len_bytes).map_err(|err| {
error!(err = ?err, id = ?id, "Failed to read loaded log hash and len");
err
}).ok()
}))
})
.filter_map(move |hash_and_len| {
let storage = storage.clone_ref();
let key = key.clone();
async move {
let all_items = storage
.get_stream(&key)
.map(|data|litl::from_slice::<litl::Val>(&data).unwrap())
.collect::<Vec<_>>()
.await
;
if all_items.len() < hash_and_len.len {
error!(id = ?id, "Storage backend returned less data than last hash");
None
} else {
Some(LogDiff {
id,
header: None,
after: 0,
new_entries: all_items[0..hash_and_len.len].to_vec(),
new_hash: Some(hash_and_len.hash),
})
}
}
})
};
header_stream.chain(all_chunks_stream).boxed_local()
}
fn store(
effective_diff: LogDiff,
storage: Box<dyn StorageBackend>,
) -> std::pin::Pin<Box<dyn futures::Future<Output = ()>>> {
let key = effective_diff.id.to_string();
async move {
if let Some(header) = &effective_diff.header {
storage
.set_key(&format!("header_{}", key), litl::to_vec(header).unwrap())
.await;
}
if let Some(new_hash) = &effective_diff.new_hash {
if effective_diff.new_entries.is_empty() {
return;
}
let new_len = effective_diff.after + effective_diff.new_entries.len();
for entry in effective_diff.new_entries {
storage
.append_to_stream(&key, litl::to_vec(&entry).unwrap(), None)
.await;
}
storage
.set_key(
&format!("hash_and_len_{}", key),
litl::to_vec(&StorageHashAndLen {
hash: new_hash.clone(),
len: new_len,
})
.unwrap(),
)
.await;
}
}
.boxed_local()
}
}
#[derive(Error, Debug)]
pub enum LogError {
#[error("Invalid header hash")]
InvalidHeaderHash,
#[error("Invalid hash after append")]
InvalidHash,
#[error(transparent)]
InvalidSignature(#[from] SignatureError),
#[error("Append out of order")]
AppendOutOfOrder,
}