Ledger workers

bee-api/bee-rest-api/Cargo.toml

@@ -11,7 +11,7 @@ keywords = ["iota", "bee", "framework", "node", "api"]
 homepage = "https://www.iota.org"
 
 [dependencies]
-bee-ledger = { version = "0.2.0", path = "../../bee-ledger" }
+bee-ledger = { version = "0.3.0", path = "../../bee-ledger" }
 bee-message = { version = "0.1.3", path = "../../bee-message" }
 bee-protocol = { version = "0.1.0", path = "../../bee-protocol" }
 

bee-ledger/CHANGELOG.md

@@ -19,7 +19,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Security -->
 
-## 0.3.0 - 2021-05-10
+## 0.3.0 - 2021-05-12
+
+### Added
+
+- `SnapshotWorker` and associated types and operations;
+- `ConsensusWorker` (White Flag) and associated types and operations;
+- `BalanceDiffs::{negate, negated, output_add, output_sub}`;
+- `impl core::fmt::{Debug, Display} for Unspent`;
 
 ### Removed
 

bee-ledger/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bee-ledger"
-version = "0.2.0"
+version = "0.3.0"
 authors = ["IOTA Stiftung"]
 edition = "2018"
 description = "All types and features required to compute and maintain the ledger state"
@@ -12,7 +12,40 @@ homepage = "https://www.iota.org"
 
 [dependencies]
 bee-common = { version = "0.4.1", path = "../bee-common/bee-common" }
-bee-message = { version = "0.1.3", path = "../bee-message" }
+bee-message = { version = "0.1.5", path = "../bee-message" }
+bee-runtime = { version = "0.1.1-alpha", path = "../bee-runtime", optional = true }
+bee-storage = { version = "0.3.0", path = "../bee-storage/bee-storage", optional = true }
+bee-tangle = { version = "0.1.1", path = "../bee-tangle", optional = true }
+bee-ternary = { version = "0.4.2-alpha", path = "../bee-ternary", optional = true }
 
-serde = { version = "1.0", optional = true }
+async-trait = { version = "0.1", optional = true }
+chrono = { version = "0.4", optional = true }
+digest = { version = "0.9", optional = true }
+futures = { version = "0.3", optional = true }
+hex = { version = "0.4", optional = true }
+iota-crypto = { version = "0.5.0", features = ["blake2b"], optional = true }
+log = { version = "0.4", optional = true }
+reqwest = { version = "0.11", features = ["stream"], optional = true }
+serde = { version = "1.0", features = ["derive" ], optional = true }
 thiserror = { version = "1.0" }
+tokio = { version = "1.4", features = ["sync", "fs"], optional = true }
+tokio-stream = { version = "0.1", optional = true }
+
+[features]
+workers = [
+  "bee-runtime",
+  "bee-storage",
+  "bee-tangle",
+  "bee-ternary",
+  "async-trait",
+  "chrono",
+  "digest",
+  "futures",
+  "hex",
+  "iota-crypto",
+  "log",
+  "reqwest",
+  "serde",
+  "tokio",
+  "tokio-stream"
+]

bee-ledger/src/lib.rs

@@ -3,6 +3,8 @@
 
 //! A crate that contains all types and features required to compute and maintain the ledger state.
 
-#![deny(missing_docs, warnings)]
+#![deny(missing_docs)]
 
 pub mod types;
+#[cfg(feature = "workers")]
+pub mod workers;

bee-ledger/src/types/balance_diff.rs

@@ -3,7 +3,11 @@
 
 use crate::types::Error;
 
-use bee_message::{address::Address, constants::IOTA_SUPPLY};
+use bee_message::{
+    address::Address,
+    constants::IOTA_SUPPLY,
+    output::{Output, DUST_THRESHOLD},
+};
 
 use std::collections::{
     hash_map::{IntoIter, Iter, IterMut},
@@ -100,6 +104,58 @@ impl BalanceDiffs {
         self.0.get(address)
     }
 
+    /// Negates a `BalanceDiffs`.
+    pub fn negate(&mut self) {
+        for (_, diff) in self.iter_mut() {
+            diff.negate();
+        }
+    }
+
+    /// Creates a new negated version of a `BalanceDiffs`.
+    pub fn negated(&self) -> Self {
+        let mut new = self.clone();
+        new.negate();
+        new
+    }
+
+    /// Adds an output to a `BalanceDiffs`.
+    pub fn output_add(&mut self, output: &Output) -> Result<(), Error> {
+        match output {
+            Output::SignatureLockedSingle(output) => {
+                self.amount_add(*output.address(), output.amount())?;
+                if output.amount() < DUST_THRESHOLD {
+                    self.dust_outputs_inc(*output.address())?;
+                }
+            }
+            Output::SignatureLockedDustAllowance(output) => {
+                self.amount_add(*output.address(), output.amount())?;
+                self.dust_allowance_add(*output.address(), output.amount())?;
+            }
+            Output::Treasury(_) => return Err(Error::UnsupportedOutputKind(output.kind())),
+        }
+
+        Ok(())
+    }
+
+    /// Subtracts an output from a BalanceDiffs`.
+    pub fn output_sub(&mut self, output: &Output) -> Result<(), Error> {
+        match output {
+            Output::SignatureLockedSingle(output) => {
+                self.amount_sub(*output.address(), output.amount())?;
+                if output.amount() < DUST_THRESHOLD {
+                    self.dust_outputs_dec(*output.address())?;
+                }
+            }
+            Output::SignatureLockedDustAllowance(output) => {
+                self.amount_sub(*output.address(), output.amount())?;
+                self.dust_allowance_sub(*output.address(), output.amount())?;
+            }
+            Output::Treasury(_) => return Err(Error::UnsupportedOutputKind(output.kind())),
+        }
+
+        Ok(())
+    }
+
     /// Adds a given amount to a given address.
     pub fn amount_add(&mut self, address: Address, amount: u64) -> Result<(), Error> {
         let entry = self.0.entry(address).or_default();
@@ -169,13 +225,6 @@ impl BalanceDiffs {
     pub fn iter_mut(&mut self) -> IterMut<'_, Address, BalanceDiff> {
         self.0.iter_mut()
     }
-
-    /// Negates a `BalanceDiffs`.
-    pub fn negate(&mut self) {
-        for (_, diff) in self.iter_mut() {
-            diff.negate();
-        }
-    }
 }
 
 impl IntoIterator for BalanceDiffs {

bee-ledger/src/types/receipt.rs

@@ -40,7 +40,9 @@ impl Receipt {
         let mut migrated_amount: u64 = 0;
         let transaction = match self.inner().transaction() {
             Payload::TreasuryTransaction(transaction) => transaction,
-            payload => return Err(Error::UnsupportedPayloadKind(payload.kind())),
+            Payload::Indexation(_) | Payload::Milestone(_) | Payload::Receipt(_) | Payload::Transaction(_) => {
+                return Err(Error::UnsupportedPayloadKind(self.inner().transaction().kind()));
+            }
         };
 
         for funds in self.inner().funds() {
@@ -62,12 +64,14 @@ impl Receipt {
                     ));
                 }
             }
-            input => return Err(Error::UnsupportedInputKind(input.kind())),
+            Input::Utxo(_) => return Err(Error::UnsupportedInputKind(transaction.input().kind())),
         };
 
         let created_treasury_output = match transaction.output() {
             Output::Treasury(output) => output,
-            output => return Err(Error::UnsupportedOutputKind(output.kind())),
+            Output::SignatureLockedDustAllowance(_) | Output::SignatureLockedSingle(_) => {
+                return Err(Error::UnsupportedOutputKind(transaction.output().kind()));
+            }
         };
 
         let created_amount = consumed_treasury_output

bee-ledger/src/types/snapshot/milestone_diff.rs

@@ -108,9 +108,15 @@ impl Packable for MilestoneDiff {
 
     fn unpack_inner<R: Read + ?Sized, const CHECK: bool>(reader: &mut R) -> Result<Self, Self::Error> {
         let milestone_len = u32::unpack_inner::<R, CHECK>(reader)? as usize;
-        let milestone = match Payload::unpack_inner::<R, CHECK>(reader)? {
+        let payload = Payload::unpack_inner::<R, CHECK>(reader)?;
+        let milestone = match payload {
             Payload::Milestone(milestone) => milestone,
-            payload => return Err(Error::InvalidPayloadKind(payload.kind())),
+            Payload::Indexation(_)
+            | Payload::Receipt(_)
+            | Payload::Transaction(_)
+            | Payload::TreasuryTransaction(_) => {
+                return Err(Error::InvalidPayloadKind(payload.kind()));
+            }
         };
 
         if milestone_len != milestone.packed_len() + std::mem::size_of_val(&MilestonePayload::KIND) {

bee-ledger/src/types/unspent.rs

@@ -9,7 +9,7 @@ use bee_message::output::OutputId;
 use std::ops::Deref;
 
 /// Represents an output id as unspent.
-#[derive(Debug, Eq, PartialEq)]
+#[derive(Eq, PartialEq)]
 pub struct Unspent(OutputId);
 
 impl From<OutputId> for Unspent {
@@ -38,6 +38,18 @@ impl Unspent {
     }
 }
 
+impl core::fmt::Display for Unspent {
+    fn fmt(&self, f: &mut core::fmt::Formatter) -> core::fmt::Result {
+        write!(f, "{}", *self)
+    }
+}
+
+impl core::fmt::Debug for Unspent {
+    fn fmt(&self, f: &mut core::fmt::Formatter) -> core::fmt::Result {
+        write!(f, "Unspent({})", self)
+    }
+}
+
 impl Packable for Unspent {
     type Error = Error;
 

bee-ledger/src/workers/consensus/merkle_hasher.rs

@@ -0,0 +1,103 @@
+// Copyright 2020-2021 IOTA Stiftung
+// SPDX-License-Identifier: Apache-2.0
+
+use bee_message::MessageId;
+
+use crypto::hashes::{Digest, Output};
+
+use std::marker::PhantomData;
+
+/// Leaf domain separation prefix.
+const LEAF_HASH_PREFIX: u8 = 0x00;
+/// Node domain separation prefix.
+const NODE_HASH_PREFIX: u8 = 0x01;
+
+/// Computes the largest power of two less than or equal to `n`.
+/// Undefined behaviour: 0 is not a valid value for `n`.
+fn largest_power_of_two(n: u32) -> usize {
+    1 << (32 - n.leading_zeros() - 1)
+}
+
+/// A Merkle hasher based on a digest function.
+pub(crate) struct MerkleHasher<D> {
+    marker: PhantomData<D>,
+}
+
+impl<D: Default + Digest> MerkleHasher<D> {
+    /// Creates a new Merkle hasher.
+    pub(crate) fn new() -> Self {
+        Self { marker: PhantomData }
+    }
+
+    /// Returns the digest of the empty hash.
+    fn empty(&mut self) -> Output<D> {
+        D::digest(&[])
+    }
+
+    /// Returns the digest of a Merkle leaf.
+    fn leaf(&mut self, message_id: MessageId) -> Output<D> {
+        let mut hasher = D::default();
+
+        hasher.update([LEAF_HASH_PREFIX]);
+        hasher.update(message_id);
+        hasher.finalize()
+    }
+
+    /// Returns the digest of a Merkle node.
+    fn node(&mut self, message_ids: &[MessageId]) -> Output<D> {
+        let mut hasher = D::default();
+        let (left, right) = message_ids.split_at(largest_power_of_two(message_ids.len() as u32 - 1));
+
+        hasher.update([NODE_HASH_PREFIX]);
+        hasher.update(self.digest_inner(left));
+        hasher.update(self.digest_inner(right));
+        hasher.finalize()
+    }
+
+    /// Returns the digest of a list of hashes as an `Output<D>`.
+    fn digest_inner(&mut self, message_ids: &[MessageId]) -> Output<D> {
+        match message_ids {
+            [] => self.empty(),
+            [message_id] => self.leaf(*message_id),
+            _ => self.node(message_ids),
+        }
+    }
+
+    /// Returns the digest of a list of hashes as a `Vec<u8>`.
+    pub(crate) fn digest(&mut self, message_ids: &[MessageId]) -> Vec<u8> {
+        self.digest_inner(message_ids).to_vec()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use super::*;
+
+    use crypto::hashes::blake2b::Blake2b256;
+
+    use std::str::FromStr;
+
+    #[test]
+    fn tree() {
+        let hashes = [
+            "52fdfc072182654f163f5f0f9a621d729566c74d10037c4d7bbb0407d1e2c649",
+            "81855ad8681d0d86d1e91e00167939cb6694d2c422acd208a0072939487f6999",
+            "eb9d18a44784045d87f3c67cf22746e995af5a25367951baa2ff6cd471c483f1",
+            "5fb90badb37c5821b6d95526a41a9504680b4e7c8b763a1b1d49d4955c848621",
+            "6325253fec738dd7a9e28bf921119c160f0702448615bbda08313f6a8eb668d2",
+            "0bf5059875921e668a5bdf2c7fc4844592d2572bcd0668d2d6c52f5054e2d083",
+            "6bf84c7174cb7476364cc3dbd968b0f7172ed85794bb358b0c3b525da1786f9f",
+        ]
+        .iter()
+        .map(|hash| MessageId::from_str(hash).unwrap())
+        .collect::<Vec<_>>();
+
+        let hash = MerkleHasher::<Blake2b256>::new().digest(&hashes);
+
+        assert_eq!(
+            hex::encode(hash),
+            "bf67ce7ba23e8c0951b5abaec4f5524360d2c26d971ff226d3359fa70cdb0beb"
+        )
+    }
+}