Minimal radicle node implementation

Signed-off-by: Alexis Sellier <self@cloudhead.io>

Cargo.toml

@@ -1,7 +1,20 @@
-[package]
-name = "radicle-node"
-version = "0.1.0"
-authors = ["Alexis Sellier <alexis@radicle.xyz>"]
-edition = "2021"
+[workspace]
+members = ["node"]
 
-[dependencies]
+[patch.crates-io.radicle-git-ext]
+git = "https://github.com/radicle-dev/radicle-link"
+tag = "cycle/2022-07-12"
+
+[patch.crates-io.git-ref-format]
+git = "https://github.com/radicle-dev/radicle-link"
+tag = "cycle/2022-07-12"
+
+[patch.crates-io.nakamoto-net]
+git = "https://github.com/cloudhead/nakamoto"
+branch = "master"
+version = "0.3.0"
+
+[patch.crates-io.nakamoto-net-poll]
+git = "https://github.com/cloudhead/nakamoto"
+branch = "master"
+version = "0.3.0"

node/Cargo.toml

@@ -0,0 +1,34 @@
+[package]
+name = "radicle-node"
+license = "MIT OR Apache-2.0"
+version = "0.2.0"
+authors = ["Alexis Sellier <alexis@radicle.xyz>"]
+edition = "2021"
+
+[dependencies]
+anyhow = { version = "1" }
+bs58 = { version = "0.4.0" }
+ed25519-consensus = { version = "2.0.1" }
+chrono = { version = "0.4.0" }
+colored = { version = "1.9.0" }
+fastrand = { version = "1.8.0" }
+git-ref-format = { version = "0", features = ["serde", "macro"] }
+git2 = { version = "0.13" }
+multibase = { version = "0.9.1" }
+log = { version = "0.4.17", features = ["std"] }
+once_cell = { version = "1.13" }
+sha2 = { version = "0.10.2" }
+serde = { version = "1", features = ["derive"] }
+serde_json = { version = "1", features = ["preserve_order"] }
+siphasher = { version = "0.3.10" }
+radicle-git-ext = { version = "0", features = ["serde"] }
+nonempty = { version = "0.8.0", features = ["serialize"] }
+nakamoto-net = { version = "0.3.0" }
+nakamoto-net-poll = { version = "0.3.0" }
+thiserror = { version = "1" }
+toml = { version = "0.5.9" }
+
+[dev-dependencies]
+quickcheck = { version = "1", default-features = false }
+quickcheck_macros = { version = "1", default-features = false }
+tempfile = { version = "3.3.0" }

node/src/address_book.rs

@@ -0,0 +1,468 @@
+use std::io::Seek;
+use std::ops::{Deref, DerefMut};
+use std::path::Path;
+use std::{fs, io, net};
+
+use crate::collections::HashMap;
+use crate::LocalTime;
+use nonempty::NonEmpty;
+use serde::{Deserialize, Serialize};
+
+/// A map with the ability to randomly select values.
+#[derive(Debug)]
+pub struct AddressBook<K, V> {
+    inner: HashMap<K, V>,
+    rng: fastrand::Rng,
+}
+
+impl<K, V> AddressBook<K, V> {
+    /// Create a new address book.
+    pub fn new(rng: fastrand::Rng) -> Self {
+        Self {
+            inner: HashMap::with_hasher(rng.clone().into()),
+            rng,
+        }
+    }
+
+    /// Pick a random value in the book.
+    pub fn sample(&self) -> Option<(&K, &V)> {
+        self.sample_with(|_, _| true)
+    }
+
+    /// Pick a random value in the book matching a predicate.
+    pub fn sample_with(&self, mut predicate: impl FnMut(&K, &V) -> bool) -> Option<(&K, &V)> {
+        if let Some(pairs) = NonEmpty::from_vec(
+            self.inner
+                .iter()
+                .filter(|(k, v)| predicate(*k, *v))
+                .collect(),
+        ) {
+            let ix = self.rng.usize(..pairs.len());
+            let pair = pairs[ix]; // Can't fail.
+
+            Some(pair)
+        } else {
+            None
+        }
+    }
+
+    /// Cycle through the keys at random. The random cycle repeats ad-infintum.
+    pub fn cycle(&self) -> impl Iterator<Item = &K> {
+        self.shuffled().map(|(k, _)| k).cycle()
+    }
+
+    /// Return a shuffled iterator over the keys.
+    pub fn shuffled(&self) -> std::vec::IntoIter<(&K, &V)> {
+        let mut keys = self.inner.iter().collect::<Vec<_>>();
+        self.rng.shuffle(&mut keys);
+
+        keys.into_iter()
+    }
+}
+
+impl<K, V> Deref for AddressBook<K, V> {
+    type Target = HashMap<K, V>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.inner
+    }
+}
+
+impl<K, V> DerefMut for AddressBook<K, V> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.inner
+    }
+}
+
+/// A known address.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct KnownAddress {
+    /// Network address.
+    pub addr: net::SocketAddr,
+    /// Address of the peer who sent us this address.
+    pub source: Source,
+    /// Last time this address was used to successfully connect to a peer.
+    #[serde(with = "local_time")]
+    pub last_success: Option<LocalTime>,
+    /// Last time this address was sampled.
+    #[serde(with = "local_time")]
+    pub last_sampled: Option<LocalTime>,
+    /// Last time this address was tried.
+    #[serde(with = "local_time")]
+    pub last_attempt: Option<LocalTime>,
+    /// Last time this peer was seen alive.
+    #[serde(with = "local_time")]
+    pub last_active: Option<LocalTime>,
+}
+
+impl KnownAddress {
+    /// Create a new known address.
+    pub fn new(addr: net::SocketAddr, source: Source, last_active: Option<LocalTime>) -> Self {
+        Self {
+            addr,
+            source,
+            last_success: None,
+            last_attempt: None,
+            last_sampled: None,
+            last_active,
+        }
+    }
+}
+
+/// Address source. Specifies where an address originated from.
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum Source {
+    /// An address that was shared by another peer.
+    Peer(net::SocketAddr),
+    /// An address that came from a DNS seed.
+    Dns,
+    /// An address that came from some source external to the system, eg.
+    /// specified by the user or added directly to the address manager.
+    Imported,
+}
+
+impl std::fmt::Display for Source {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Peer(addr) => write!(f, "{}", addr),
+            Self::Dns => write!(f, "DNS"),
+            Self::Imported => write!(f, "Imported"),
+        }
+    }
+}
+
+/// A file-backed address cache.
+#[derive(Debug)]
+pub struct Cache {
+    addrs: std::collections::HashMap<net::IpAddr, KnownAddress>,
+    file: fs::File,
+}
+
+impl Cache {
+    /// Open an existing cache.
+    pub fn open<P: AsRef<Path>>(path: P) -> io::Result<Self> {
+        fs::OpenOptions::new()
+            .read(true)
+            .write(true)
+            .open(path)
+            .and_then(Self::from)
+    }
+
+    /// Create a new cache.
+    pub fn create<P: AsRef<Path>>(path: P) -> io::Result<Self> {
+        use std::collections::HashMap;
+
+        let file = fs::OpenOptions::new()
+            .create_new(true)
+            .write(true)
+            .open(path)?;
+
+        Ok(Self {
+            file,
+            addrs: HashMap::new(),
+        })
+    }
+
+    /// Create a new cache from a file.
+    pub fn from(mut file: fs::File) -> io::Result<Self> {
+        use std::collections::HashMap;
+
+        let bytes = file.seek(io::SeekFrom::End(0))?;
+        let addrs = if bytes == 0 {
+            HashMap::new()
+        } else {
+            file.rewind()?;
+            serde_json::from_reader(&file)?
+        };
+
+        Ok(Self { file, addrs })
+    }
+}
+
+impl Store for Cache {
+    fn get_mut(&mut self, ip: &net::IpAddr) -> Option<&mut KnownAddress> {
+        self.addrs.get_mut(ip)
+    }
+
+    fn get(&self, ip: &net::IpAddr) -> Option<&KnownAddress> {
+        self.addrs.get(ip)
+    }
+
+    fn remove(&mut self, ip: &net::IpAddr) -> Option<KnownAddress> {
+        self.addrs.remove(ip)
+    }
+
+    fn insert(&mut self, ip: net::IpAddr, ka: KnownAddress) -> bool {
+        <std::collections::HashMap<_, _> as Store>::insert(&mut self.addrs, ip, ka)
+    }
+
+    fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = (&net::IpAddr, &KnownAddress)> + 'a> {
+        Box::new(self.addrs.iter())
+    }
+
+    fn clear(&mut self) {
+        self.addrs.clear()
+    }
+
+    fn len(&self) -> usize {
+        self.addrs.len()
+    }
+
+    fn flush<'a>(&mut self) -> io::Result<()> {
+        use io::Write;
+
+        let peers = serde_json::to_value(&self.addrs)?;
+        let s = serde_json::to_string(&peers)?;
+
+        self.file.set_len(0)?;
+        self.file.seek(io::SeekFrom::Start(0))?;
+        self.file.write_all(s.as_bytes())?;
+        self.file.write_all(&[b'\n'])?;
+        self.file.sync_data()?;
+
+        Ok(())
+    }
+}
+
+/// Address store.
+///
+/// Used to store peer addresses and metadata.
+pub trait Store {
+    /// Get a known peer address.
+    fn get(&self, ip: &net::IpAddr) -> Option<&KnownAddress>;
+
+    /// Get a known peer address mutably.
+    fn get_mut(&mut self, ip: &net::IpAddr) -> Option<&mut KnownAddress>;
+
+    /// Insert a *new* address into the store. Returns `true` if the address was inserted,
+    /// or `false` if it was already known.
+    fn insert(&mut self, ip: net::IpAddr, ka: KnownAddress) -> bool;
+
+    /// Remove an address from the store.
+    fn remove(&mut self, ip: &net::IpAddr) -> Option<KnownAddress>;
+
+    /// Return an iterator over the known addresses.
+    fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = (&net::IpAddr, &KnownAddress)> + 'a>;
+
+    /// Returns the number of addresses.
+    fn len(&self) -> usize;
+
+    /// Returns true if there are no addresses.
+    fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    /// Seed the peer store with addresses.
+    /// Fails if *none* of the seeds could be resolved to addresses.
+    fn seed<S: net::ToSocketAddrs>(
+        &mut self,
+        seeds: impl Iterator<Item = S>,
+        source: Source,
+    ) -> io::Result<()> {
+        let mut error = None;
+        let mut success = false;
+
+        for seed in seeds {
+            match seed.to_socket_addrs() {
+                Ok(addrs) => {
+                    success = true;
+                    for addr in addrs {
+                        self.insert(addr.ip(), KnownAddress::new(addr, source, None));
+                    }
+                }
+                Err(err) => error = Some(err),
+            }
+        }
+
+        if success {
+            return Ok(());
+        }
+        if let Some(err) = error {
+            return Err(io::Error::new(
+                io::ErrorKind::Other,
+                format!("seeds failed to resolve: {}", err),
+            ));
+        }
+        Ok(())
+    }
+
+    /// Clears the store of all addresses.
+    fn clear(&mut self);
+
+    /// Flush data to permanent storage.
+    fn flush(&mut self) -> io::Result<()>;
+}
+
+/// Implementation of [`Store`] for [`std::collections::HashMap`].
+impl Store for std::collections::HashMap<net::IpAddr, KnownAddress> {
+    fn get_mut(&mut self, ip: &net::IpAddr) -> Option<&mut KnownAddress> {
+        self.get_mut(ip)
+    }
+
+    fn get(&self, ip: &net::IpAddr) -> Option<&KnownAddress> {
+        self.get(ip)
+    }
+
+    fn remove(&mut self, ip: &net::IpAddr) -> Option<KnownAddress> {
+        self.remove(ip)
+    }
+
+    fn insert(&mut self, ip: net::IpAddr, ka: KnownAddress) -> bool {
+        use std::collections::hash_map::Entry;
+
+        match self.entry(ip) {
+            Entry::Vacant(v) => {
+                v.insert(ka);
+            }
+            Entry::Occupied(_) => return false,
+        }
+        true
+    }
+
+    fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = (&net::IpAddr, &KnownAddress)> + 'a> {
+        Box::new(self.iter())
+    }
+
+    fn clear(&mut self) {
+        self.clear()
+    }
+
+    fn len(&self) -> usize {
+        self.len()
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+/// Implementation of [`Store`] for [`crate::collections::HashMap`].
+impl Store for crate::collections::HashMap<net::IpAddr, KnownAddress> {
+    fn get_mut(&mut self, ip: &net::IpAddr) -> Option<&mut KnownAddress> {
+        self.get_mut(ip)
+    }
+
+    fn get(&self, ip: &net::IpAddr) -> Option<&KnownAddress> {
+        self.get(ip)
+    }
+
+    fn remove(&mut self, ip: &net::IpAddr) -> Option<KnownAddress> {
+        self.remove(ip)
+    }
+
+    fn insert(&mut self, ip: net::IpAddr, ka: KnownAddress) -> bool {
+        use std::collections::hash_map::Entry;
+
+        match self.entry(ip) {
+            Entry::Vacant(v) => {
+                v.insert(ka);
+            }
+            Entry::Occupied(_) => return false,
+        }
+        true
+    }
+
+    fn iter<'a>(&'a self) -> Box<dyn Iterator<Item = (&net::IpAddr, &KnownAddress)> + 'a> {
+        Box::new(self.iter())
+    }
+
+    fn clear(&mut self) {
+        self.clear()
+    }
+
+    fn len(&self) -> usize {
+        self.len()
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+mod local_time {
+    use super::LocalTime;
+    use serde::{Deserialize, Deserializer, Serializer};
+
+    pub fn deserialize<'de, D>(deserializer: D) -> Result<Option<LocalTime>, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let value: Option<u64> = Deserialize::deserialize(deserializer)?;
+
+        if let Some(value) = value {
+            Ok(Some(LocalTime::from_secs(value)))
+        } else {
+            Ok(None)
+        }
+    }
+
+    pub fn serialize<S>(value: &Option<LocalTime>, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        if let Some(local_time) = value {
+            serializer.serialize_u64(local_time.as_secs())
+        } else {
+            serializer.serialize_none()
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn test_empty() {
+        let tmp = tempfile::tempdir().unwrap();
+        let path = tmp.path().join("cache");
+
+        Cache::create(&path).unwrap();
+        let cache = Cache::open(&path).unwrap();
+
+        assert!(cache.is_empty());
+    }
+
+    #[test]
+    fn test_save_and_load() {
+        let tmp = tempfile::tempdir().unwrap();
+        let path = tmp.path().join("cache");
+        let mut expected = Vec::new();
+
+        {
+            let mut cache = Cache::create(&path).unwrap();
+
+            for i in 32..48 {
+                let ip = net::IpAddr::from([127, 0, 0, i]);
+                let addr = net::SocketAddr::from((ip, 8333));
+                let ka = KnownAddress {
+                    addr,
+                    source: Source::Dns,
+                    last_success: Some(LocalTime::from_secs(i as u64)),
+                    last_sampled: Some(LocalTime::from_secs((i + 1) as u64)),
+                    last_attempt: None,
+                    last_active: None,
+                };
+                cache.insert(ip, ka);
+            }
+            cache.flush().unwrap();
+
+            for (ip, ka) in cache.iter() {
+                expected.push((*ip, ka.clone()));
+            }
+        }
+
+        {
+            let cache = Cache::open(&path).unwrap();
+            let mut actual = cache
+                .iter()
+                .map(|(i, ka)| (*i, ka.clone()))
+                .collect::<Vec<_>>();
+
+            actual.sort_by_key(|(i, _)| *i);
+            expected.sort_by_key(|(i, _)| *i);
+
+            assert_eq!(actual, expected);
+        }
+    }
+}

node/src/address_manager.rs

@@ -0,0 +1,12 @@
+use crate::address_book::Store;
+
+#[derive(Debug)]
+pub struct AddressManager<S> {
+    store: S,
+}
+
+impl<S: Store> AddressManager<S> {
+    pub fn new(store: S) -> Self {
+        Self { store }
+    }
+}

node/src/clock.rs

@@ -0,0 +1,37 @@
+use std::cell::RefCell;
+use std::rc::Rc;
+
+use crate::{LocalDuration, LocalTime};
+
+/// Clock with interior mutability.
+#[derive(Debug, Clone)]
+pub struct RefClock(Rc<RefCell<LocalTime>>);
+
+impl std::ops::Deref for RefClock {
+    type Target = Rc<RefCell<LocalTime>>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl RefClock {
+    /// Elapse time.
+    pub fn elapse(&self, duration: LocalDuration) {
+        self.borrow_mut().elapse(duration)
+    }
+
+    pub fn local_time(&self) -> LocalTime {
+        *self.borrow()
+    }
+
+    pub fn set(&mut self, time: LocalTime) {
+        *self.borrow_mut() = time;
+    }
+}
+
+impl From<LocalTime> for RefClock {
+    fn from(other: LocalTime) -> Self {
+        Self(Rc::new(RefCell::new(other)))
+    }
+}

node/src/collections.rs

@@ -0,0 +1,38 @@
+//! Useful collections for peer-to-peer networking.
+use siphasher::sip::SipHasher13;
+
+/// A `HashMap` which uses [`fastrand::Rng`] for its random state.
+pub type HashMap<K, V> = std::collections::HashMap<K, V, RandomState>;
+
+/// A `HashSet` which uses [`fastrand::Rng`] for its random state.
+pub type HashSet<K> = std::collections::HashSet<K, RandomState>;
+
+/// Random hasher state.
+#[derive(Default, Clone)]
+pub struct RandomState {
+    key1: u64,
+    key2: u64,
+}
+
+impl RandomState {
+    fn new(rng: fastrand::Rng) -> Self {
+        Self {
+            key1: rng.u64(..),
+            key2: rng.u64(..),
+        }
+    }
+}
+
+impl std::hash::BuildHasher for RandomState {
+    type Hasher = SipHasher13;
+
+    fn build_hasher(&self) -> Self::Hasher {
+        SipHasher13::new_with_keys(self.key1, self.key2)
+    }
+}
+
+impl From<fastrand::Rng> for RandomState {
+    fn from(rng: fastrand::Rng) -> Self {
+        Self::new(rng)
+    }
+}

node/src/decoder.rs

@@ -0,0 +1,104 @@
+use std::marker::PhantomData;
+
+use crate::protocol::Envelope;
+use serde::Deserialize;
+
+/// Message stream decoder.
+///
+/// Used to for example turn a byte stream into network messages.
+#[derive(Debug)]
+pub struct Decoder<D = Envelope> {
+    unparsed: Vec<u8>,
+    item: PhantomData<D>,
+}
+
+impl<D> From<Vec<u8>> for Decoder<D> {
+    fn from(unparsed: Vec<u8>) -> Self {
+        Self {
+            unparsed,
+            item: PhantomData,
+        }
+    }
+}
+
+impl<'de, D: Deserialize<'de>> Decoder<D> {
+    /// Create a new stream decoder.
+    pub fn new(capacity: usize) -> Self {
+        Self {
+            unparsed: Vec::with_capacity(capacity),
+            item: PhantomData,
+        }
+    }
+
+    /// Input bytes into the decoder.
+    pub fn input(&mut self, bytes: &[u8]) {
+        self.unparsed.extend_from_slice(bytes);
+    }
+
+    /// Decode and return the next message. Returns [`None`] if nothing was decoded.
+    pub fn decode_next(&mut self) -> Result<Option<D>, serde_json::Error> {
+        let mut de = serde_json::Deserializer::from_reader(self.unparsed.as_slice()).into_iter();
+
+        match de.next() {
+            Some(Ok(msg)) => {
+                self.unparsed.drain(..de.byte_offset());
+                Ok(Some(msg))
+            }
+            Some(Err(err)) if err.is_eof() => Ok(None),
+
+            result => result.transpose(),
+        }
+    }
+}
+
+impl<'de, D: Deserialize<'de>> Iterator for Decoder<D> {
+    type Item = Result<D, serde_json::Error>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.decode_next().transpose()
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use quickcheck_macros::quickcheck;
+
+    const MSG_HELLO: &[u8] = b"{\"cmd\":\"hello\"}";
+    const MSG_BYE: &[u8] = b"{\"cmd\":\"goodbye\"}";
+
+    #[quickcheck]
+    fn prop_decode_next(chunk_size: usize) {
+        let mut bytes = vec![];
+        let mut msgs = vec![];
+        let mut decoder = Decoder::<serde_json::Value>::new(64);
+
+        let chunk_size = 1 + chunk_size % MSG_HELLO.len() + MSG_BYE.len();
+
+        bytes.extend_from_slice(MSG_HELLO);
+        bytes.extend_from_slice(MSG_BYE);
+
+        for chunk in bytes.as_slice().chunks(chunk_size) {
+            decoder.input(chunk);
+
+            while let Some(msg) = decoder.decode_next().unwrap() {
+                msgs.push(msg);
+            }
+        }
+
+        assert_eq!(decoder.unparsed.len(), 0);
+        assert_eq!(msgs.len(), 2);
+        assert_eq!(
+            msgs[0],
+            serde_json::json!({
+                "cmd": "hello",
+            })
+        );
+        assert_eq!(
+            msgs[1],
+            serde_json::json!({
+                "cmd": "goodbye",
+            })
+        );
+    }
+}

node/src/hash.rs

@@ -0,0 +1,79 @@
+use std::fmt;
+
+use serde::{Deserialize, Serialize};
+use sha2::{
+    digest::{generic_array::GenericArray, OutputSizeUser},
+    Digest as _, Sha256,
+};
+use thiserror::Error;
+
+#[derive(Debug, Clone, PartialEq, Eq, Error)]
+pub enum ParseError {
+    #[error("invalid string length")]
+    InvalidLength,
+    #[error(transparent)]
+    ParseInt(#[from] std::num::ParseIntError),
+}
+
+/// A SHA-256 hash.
+#[derive(Serialize, Deserialize, Default, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct Digest([u8; 32]);
+
+impl Digest {
+    pub fn new(bytes: impl AsRef<[u8]>) -> Self {
+        Self::from(Sha256::digest(bytes))
+    }
+
+    pub fn encode(&self) -> String {
+        self.to_string()
+    }
+
+    pub fn decode(s: &str) -> Result<Self, ParseError> {
+        if s.len() != 64 {
+            Err(ParseError::InvalidLength)
+        } else {
+            let mut bytes: [u8; 32] = Default::default();
+            for (i, byte) in (0..s.len())
+                .step_by(2)
+                .map(|i| u8::from_str_radix(&s[i..i + 2], 16))
+                .enumerate()
+            {
+                bytes[i] = byte?;
+            }
+            Ok(Self(bytes))
+        }
+    }
+}
+
+impl AsRef<[u8]> for Digest {
+    fn as_ref(&self) -> &[u8] {
+        &self.0
+    }
+}
+
+impl fmt::Debug for Digest {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "Hash({})", self.encode())
+    }
+}
+
+impl fmt::Display for Digest {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        for byte in &self.0 {
+            write!(f, "{:02x}", byte)?;
+        }
+        Ok(())
+    }
+}
+
+impl From<[u8; 32]> for Digest {
+    fn from(bytes: [u8; 32]) -> Self {
+        Self(bytes)
+    }
+}
+
+impl From<GenericArray<u8, <Sha256 as OutputSizeUser>::OutputSize>> for Digest {
+    fn from(array: GenericArray<u8, <Sha256 as OutputSizeUser>::OutputSize>) -> Self {
+        Self(array.into())
+    }
+}

node/src/identity.rs

@@ -0,0 +1,174 @@
+use std::{fmt, io, ops::Deref, str::FromStr};
+
+use ed25519_consensus::{VerificationKey, VerificationKeyBytes};
+use nonempty::NonEmpty;
+use radicle_git_ext::Oid;
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+
+use crate::hash;
+
+#[derive(Serialize, Deserialize, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct ProjId(hash::Digest);
+
+impl fmt::Display for ProjId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.encode())
+    }
+}
+
+impl fmt::Debug for ProjId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "ProjId({})", self.encode())
+    }
+}
+
+impl ProjId {
+    pub fn encode(&self) -> String {
+        multibase::encode(multibase::Base::Base58Btc, &self.0.as_ref())
+    }
+
+    pub(crate) fn from_ref(s: &str) -> Result<ProjId, IdError> {
+        if let Some(s) = s.split('/').nth(2) {
+            let mut array: [u8; 32] = [0; 32];
+            let bytes = bs58::decode(s).into(&mut array)?;
+
+            // TODO: Multi-hash?
+
+            assert_eq!(bytes, array.len());
+
+            return Ok(Self(hash::Digest::from(array)));
+        }
+        Err(IdError::InvalidRef(s.to_owned()))
+    }
+}
+
+impl From<hash::Digest> for ProjId {
+    fn from(digest: hash::Digest) -> Self {
+        Self(digest)
+    }
+}
+
+#[derive(Serialize, Deserialize, PartialEq, Eq, Hash, Debug, Clone)]
+pub struct Did(UserId);
+
+impl Did {
+    fn encode(&self) -> String {
+        format!("did:key:{}", self.0.encode())
+    }
+}
+
+impl fmt::Display for Did {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.encode())
+    }
+}
+
+#[derive(Serialize, Deserialize, Eq, Debug, Clone)]
+pub struct UserId(pub VerificationKey);
+
+impl std::hash::Hash for UserId {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.0.as_bytes().hash(state)
+    }
+}
+
+impl PartialEq for UserId {
+    fn eq(&self, other: &Self) -> bool {
+        self.0 == other.0
+    }
+}
+
+impl UserId {
+    fn encode(&self) -> String {
+        multibase::encode(multibase::Base::Base58Btc, &self.0)
+    }
+}
+
+impl FromStr for UserId {
+    type Err = IdError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let mut array: [u8; 32] = [0; 32];
+        let bytes = bs58::decode(s).into(&mut array)?;
+        let key = VerificationKey::try_from(VerificationKeyBytes::from(array))?;
+
+        assert_eq!(bytes, array.len());
+
+        Ok(Self(key))
+    }
+}
+
+impl Deref for UserId {
+    type Target = VerificationKey;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+#[derive(Error, Debug)]
+pub enum IdError {
+    #[error("invalid ref '{0}'")]
+    InvalidRef(String),
+    #[error("invalid base58 string: {0}")]
+    Base58(#[from] bs58::decode::Error),
+    #[error("invalid key: {0}")]
+    InvalidKey(#[from] ed25519_consensus::Error),
+}
+
+impl UserId {}
+
+#[derive(Error, Debug)]
+pub enum DocError {
+    #[error("toml: {0}")]
+    Toml(#[from] toml::ser::Error),
+    #[error("i/o: {0}")]
+    Io(#[from] io::Error),
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct Delegate {
+    pub name: String,
+    pub id: Did,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct Doc {
+    pub name: String,
+    pub description: String,
+    pub version: u32,
+    pub parent: Oid,
+    pub delegate: NonEmpty<Delegate>,
+}
+
+impl Doc {
+    pub fn write<W: io::Write>(&self, mut writer: W) -> Result<ProjId, DocError> {
+        let buf = toml::to_string_pretty(self)?;
+        let digest = hash::Digest::new(buf.as_bytes());
+        let id = ProjId::from(digest);
+
+        writer.write_all(buf.as_bytes())?;
+
+        Ok(id)
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use quickcheck_macros::quickcheck;
+    use std::collections::HashSet;
+
+    #[quickcheck]
+    fn prop_user_id_equality(a: UserId, b: UserId) {
+        assert_ne!(a, b);
+
+        let mut hm = HashSet::new();
+
+        assert!(hm.insert(a.clone()));
+        assert!(hm.insert(b.clone()));
+        assert!(!hm.insert(a));
+        assert!(!hm.insert(b));
+    }
+}

node/src/lib.rs

@@ -0,0 +1,19 @@
+#![allow(dead_code)]
+pub use nakamoto_net::{Io, Link, LocalDuration, LocalTime};
+
+mod address_book;
+mod address_manager;
+mod clock;
+mod collections;
+mod decoder;
+mod hash;
+mod identity;
+mod logger;
+mod protocol;
+mod storage;
+#[cfg(test)]
+mod test;
+
+pub fn run() -> anyhow::Result<()> {
+    Ok(())
+}

node/src/logger.rs

@@ -0,0 +1,61 @@
+//! Logging module.
+use std::io;
+
+use chrono::prelude::*;
+use colored::*;
+use log::{Level, Log, Metadata, Record, SetLoggerError};
+
+struct Logger {
+    level: Level,
+}
+
+impl Log for Logger {
+    fn enabled(&self, metadata: &Metadata) -> bool {
+        metadata.level() <= self.level
+    }
+
+    fn log(&self, record: &Record) {
+        if self.enabled(record.metadata()) {
+            let module = record.module_path().unwrap_or_default();
+
+            if record.level() == Level::Error {
+                write(record, module, io::stderr());
+            } else {
+                write(record, module, io::stdout());
+            }
+
+            fn write(record: &log::Record, module: &str, mut stream: impl io::Write) {
+                let message = format!("{} {} {}", record.level(), module.bold(), record.args());
+                let message = match record.level() {
+                    Level::Error => message.red(),
+                    Level::Warn => message.yellow(),
+                    Level::Info => message.normal(),
+                    Level::Debug => message.dimmed(),
+                    Level::Trace => message.white().dimmed(),
+                };
+
+                writeln!(
+                    stream,
+                    "{} {}",
+                    Local::now()
+                        .to_rfc3339_opts(SecondsFormat::Millis, true)
+                        .white(),
+                    message,
+                )
+                .expect("write shouldn't fail");
+            }
+        }
+    }
+
+    fn flush(&self) {}
+}
+
+/// Initialize a new logger.
+pub fn init(level: Level) -> Result<(), SetLoggerError> {
+    let logger = Logger { level };
+
+    log::set_boxed_logger(Box::new(logger))?;
+    log::set_max_level(level.to_level_filter());
+
+    Ok(())
+}

node/src/main.rs

@@ -0,0 +1,3 @@
+fn main() -> anyhow::Result<()> {
+    radicle_node::run()
+}

node/src/protocol.rs

@@ -0,0 +1,806 @@
+#![allow(dead_code)]
+use std::ops::{Deref, DerefMut};
+use std::{collections::VecDeque, fmt, io, net, net::IpAddr};
+
+use fastrand::Rng;
+use log::*;
+use nakamoto::{LocalDuration, LocalTime};
+use nakamoto_net as nakamoto;
+use nakamoto_net::{Io, Link};
+use serde::{Deserialize, Serialize};
+
+use crate::address_book;
+use crate::address_book::AddressBook;
+use crate::address_manager::AddressManager;
+use crate::clock::RefClock;
+use crate::collections::{HashMap, HashSet};
+use crate::decoder::Decoder;
+use crate::identity::{ProjId, UserId};
+use crate::storage;
+use crate::storage::{Inventory, ReadStorage, Refs, WriteStorage};
+
+/// Network peer identifier.
+pub type PeerId = IpAddr;
+/// Network routing table. Keeps track of where projects are hosted.
+pub type Routing = HashMap<ProjId, HashSet<PeerId>>;
+
+pub const NETWORK_MAGIC: u32 = 0x819b43d9;
+pub const DEFAULT_PORT: u16 = 8776;
+pub const PROTOCOL_VERSION: u32 = 1;
+pub const TARGET_OUTBOUND_PEERS: usize = 8;
+pub const IDLE_INTERVAL: LocalDuration = LocalDuration::from_secs(30);
+pub const ANNOUNCE_INTERVAL: LocalDuration = LocalDuration::from_secs(30);
+pub const SYNC_INTERVAL: LocalDuration = LocalDuration::from_secs(60);
+pub const PRUNE_INTERVAL: LocalDuration = LocalDuration::from_mins(30);
+pub const MAX_CONNECTION_ATTEMPTS: usize = 3;
+
+/// Commands sent to the protocol by the operator.
+#[derive(Debug)]
+pub enum Command {
+    Connect(net::SocketAddr),
+}
+
+/// Message envelope. All messages sent over the network are wrapped in this type.
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Envelope {
+    /// Network magic constant. Used to differentiate networks.
+    pub magic: u32,
+    /// The message payload.
+    pub msg: Message,
+}
+
+/// Message payload.
+/// These are the messages peers send to each other.
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub enum Message {
+    /// Say hello to a peer. This is the first message sent to a peer after connection.
+    Hello { version: u32 },
+    /// Get node addresses from a peer.
+    GetAddrs,
+    /// Send node addresses to a peer. Sent in response to [`Message::GetAddrs`].
+    Addrs { addrs: Vec<net::SocketAddr> },
+    /// Get a peer's inventory.
+    GetInventory { ids: Vec<ProjId> },
+    /// Send our inventory to a peer. Sent in response to [`Message::GetInventory`].
+    /// Nb. This should be the whole inventory, not a partial update.
+    Inventory { seq: u64, inv: Inventory },
+}
+
+impl From<Message> for Envelope {
+    fn from(msg: Message) -> Self {
+        Self {
+            magic: NETWORK_MAGIC,
+            msg,
+        }
+    }
+}
+
+impl Message {
+    pub fn hello() -> Self {
+        Self::Hello {
+            version: PROTOCOL_VERSION,
+        }
+    }
+
+    pub fn inventory<S, T>(ctx: &mut Context<S, T>) -> Result<Self, storage::Error>
+    where
+        T: storage::ReadStorage,
+    {
+        ctx.seq += 1;
+
+        let seq = ctx.seq;
+        let inv = ctx.storage.inventory()?;
+
+        Ok(Self::Inventory { seq, inv })
+    }
+
+    pub fn get_inventory(ids: impl Into<Vec<ProjId>>) -> Self {
+        Self::GetInventory { ids: ids.into() }
+    }
+}
+
+/// Project tracking policy.
+#[derive(Debug)]
+pub enum ProjectTracking {
+    /// Track all projects we come across.
+    All { blocked: HashSet<ProjId> },
+    /// Track a static list of projects.
+    Allowed(HashSet<ProjId>),
+}
+
+impl Default for ProjectTracking {
+    fn default() -> Self {
+        Self::All {
+            blocked: HashSet::default(),
+        }
+    }
+}
+
+/// Project remote tracking policy.
+#[derive(Debug, Default)]
+pub enum RemoteTracking {
+    /// Only track remotes of project delegates.
+    #[default]
+    DelegatesOnly,
+    /// Track all remotes.
+    All { blocked: HashSet<UserId> },
+    /// Track a specific list of users as well as the project delegates.
+    Allowed(HashSet<UserId>),
+}
+
+/// Protocol configuration.
+#[derive(Debug, Default)]
+pub struct Config {
+    /// Peers to connect to on startup.
+    /// Connections to these peers will be maintained.
+    pub connect: Vec<net::SocketAddr>,
+    /// Project tracking policy.
+    pub project_tracking: ProjectTracking,
+    /// Project remote tracking policy.
+    pub remote_tracking: RemoteTracking,
+}
+
+impl Config {
+    pub fn is_persistent(&self, addr: &net::SocketAddr) -> bool {
+        self.connect.contains(addr)
+    }
+
+    /// Track a project. Returns whether the policy was updated.
+    pub fn track(&mut self, proj: ProjId) -> bool {
+        match &mut self.project_tracking {
+            ProjectTracking::All { .. } => false,
+            ProjectTracking::Allowed(projs) => projs.insert(proj),
+        }
+    }
+
+    /// Untrack a project. Returns whether the policy was updated.
+    pub fn untrack(&mut self, proj: ProjId) -> bool {
+        match &mut self.project_tracking {
+            ProjectTracking::All { blocked } => blocked.insert(proj),
+            ProjectTracking::Allowed(projs) => projs.remove(&proj),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct Protocol<S, T> {
+    /// Peers currently or recently connected.
+    peers: Peers,
+    /// Protocol state that isn't peer-specific.
+    context: Context<S, T>,
+    /// Whether our local inventory no long represents what we have announced to the network.
+    out_of_sync: bool,
+    /// Last time the protocol was idle.
+    last_idle: LocalTime,
+    /// Last time the protocol synced.
+    last_sync: LocalTime,
+    /// Last time the protocol routing table was pruned.
+    last_prune: LocalTime,
+    /// Last time the protocol announced its inventory.
+    last_announce: LocalTime,
+    /// Time when the protocol was initialized.
+    start_time: LocalTime,
+}
+
+impl<T: ReadStorage + WriteStorage, S: address_book::Store> Protocol<S, T> {
+    pub fn new(config: Config, clock: RefClock, storage: T, addresses: S, rng: Rng) -> Self {
+        let addrmgr = AddressManager::new(addresses);
+
+        Self {
+            context: Context::new(config, clock, storage, addrmgr, rng.clone()),
+            peers: Peers::new(rng),
+            out_of_sync: false,
+            last_idle: LocalTime::default(),
+            last_sync: LocalTime::default(),
+            last_prune: LocalTime::default(),
+            last_announce: LocalTime::default(),
+            start_time: LocalTime::default(),
+        }
+    }
+
+    pub fn disconnect(&mut self, peer: &PeerId) {
+        if let Some(addr) = self.peers.get(peer).map(|p| p.addr) {
+            self.outbox()
+                .push_back(Io::Disconnect(addr, DisconnectReason::User));
+        }
+    }
+
+    pub fn providers(&self, proj: &ProjId) -> Box<dyn Iterator<Item = &Peer> + '_> {
+        if let Some(peers) = self.routing.get(proj) {
+            Box::new(peers.iter().filter_map(|id| self.peers.get(id)))
+        } else {
+            Box::new(std::iter::empty())
+        }
+    }
+
+    pub fn tracked(&self) -> Result<Vec<ProjId>, storage::Error> {
+        let tracked = match &self.config.project_tracking {
+            ProjectTracking::All { blocked } => self
+                .storage
+                .inventory()?
+                .into_iter()
+                .map(|(id, _)| id)
+                .filter(|id| !blocked.contains(id))
+                .collect(),
+
+            ProjectTracking::Allowed(projs) => projs.iter().cloned().collect(),
+        };
+
+        Ok(tracked)
+    }
+
+    /// Track a project.
+    /// Returns whether or not the tracking policy was updated.
+    pub fn track(&mut self, proj: ProjId) -> bool {
+        self.out_of_sync = self.config.track(proj);
+        self.out_of_sync
+    }
+
+    /// Untrack a project.
+    /// Returns whether or not the tracking policy was updated.
+    /// Note that when untracking, we don't announce anything to the network. This is because by
+    /// simply not announcing it anymore, it will eventually be pruned by nodes.
+    pub fn untrack(&mut self, proj: ProjId) -> bool {
+        self.config.untrack(proj)
+    }
+
+    /// Find the closest `n` peers by proximity in tracking graphs.
+    /// Returns a sorted list from the closest peer to the furthest.
+    /// Peers with more trackings in common score score higher.
+    #[allow(unused)]
+    pub fn closest_peers(&self, n: usize) -> Vec<PeerId> {
+        todo!()
+    }
+
+    /// Get the connected peers.
+    pub fn peers(&self) -> &Peers {
+        &self.peers
+    }
+
+    /// Get the current inventory.
+    pub fn inventory(&self) -> Result<Inventory, storage::Error> {
+        self.context.storage.inventory()
+    }
+
+    /// Get the storage instance.
+    pub fn storage(&self) -> &T {
+        &self.context.storage
+    }
+
+    /// Get the local protocol time.
+    pub fn local_time(&self) -> LocalTime {
+        self.context.clock.local_time()
+    }
+
+    /// Get protocol configuration.
+    pub fn config(&self) -> &Config {
+        &self.context.config
+    }
+
+    /// Get reference to routing table.
+    pub fn routing(&self) -> &Routing {
+        &self.context.routing
+    }
+
+    /// Get I/O outbox.
+    pub fn outbox(&mut self) -> &mut VecDeque<Io<(), DisconnectReason>> {
+        &mut self.context.io
+    }
+
+    pub fn lookup(&self, proj: &ProjId) -> Lookup {
+        Lookup {
+            local: self.context.storage.get(proj).unwrap(),
+            remote: self
+                .context
+                .routing
+                .get(proj)
+                .map_or(vec![], |r| r.iter().copied().collect()),
+        }
+    }
+
+    ////////////////////////////////////////////////////////////////////////////
+    // Periodic tasks
+    ////////////////////////////////////////////////////////////////////////////
+
+    /// Announce our inventory to all connected peers.
+    fn announce_inventory(&mut self) -> Result<(), storage::Error> {
+        let inv = Message::inventory(&mut self.context)?;
+
+        for addr in self.peers.negotiated().map(|(_, p)| p.addr) {
+            self.context.write(addr, inv.clone());
+        }
+        Ok(())
+    }
+
+    fn get_inventories(&mut self) -> Result<(), storage::Error> {
+        let mut msgs = Vec::new();
+        for proj in self.tracked()? {
+            for peer in self.providers(&proj) {
+                if peer.is_negotiated() {
+                    msgs.push((
+                        peer.addr,
+                        Message::GetInventory {
+                            ids: vec![proj.clone()],
+                        },
+                    ));
+                }
+            }
+        }
+        for (remote, msg) in msgs {
+            self.write(remote, msg);
+        }
+
+        Ok(())
+    }
+
+    fn prune_routing_entries(&mut self) {
+        // TODO
+    }
+
+    fn maintain_connections(&mut self) {
+        // TODO: Connect to all potential providers.
+        if self.peers.len() < TARGET_OUTBOUND_PEERS {
+            let delta = TARGET_OUTBOUND_PEERS - self.peers.len();
+
+            for _ in 0..delta {
+                // TODO: Connect to random peer.
+            }
+        }
+    }
+}
+
+impl<S, T> nakamoto::Protocol for Protocol<S, T>
+where
+    T: ReadStorage + WriteStorage,
+    S: address_book::Store,
+{
+    type Event = ();
+    type Command = Command;
+    type DisconnectReason = DisconnectReason;
+
+    fn initialize(&mut self, time: LocalTime) {
+        trace!("Init {}", time.as_secs());
+
+        self.start_time = time;
+
+        // Connect to configured peers.
+        let addrs = self.context.config.connect.clone();
+        for addr in addrs {
+            self.context.connect(addr);
+        }
+    }
+
+    fn tick(&mut self, now: nakamoto::LocalTime) {
+        trace!("Tick +{}", now - self.start_time);
+
+        self.context.clock.set(now);
+    }
+
+    fn wake(&mut self) {
+        let now = self.context.clock.local_time();
+
+        trace!("Wake +{}", now - self.start_time);
+
+        if now - self.last_idle >= IDLE_INTERVAL {
+            debug!("Running 'idle' task...");
+
+            self.maintain_connections();
+            self.context.io.push_back(Io::Wakeup(IDLE_INTERVAL));
+            self.last_idle = now;
+        }
+        if now - self.last_sync >= SYNC_INTERVAL {
+            debug!("Running 'sync' task...");
+
+            self.get_inventories().unwrap();
+            self.context.io.push_back(Io::Wakeup(SYNC_INTERVAL));
+            self.last_sync = now;
+        }
+        if now - self.last_announce >= ANNOUNCE_INTERVAL {
+            if self.out_of_sync {
+                self.announce_inventory().unwrap();
+            }
+            self.context.io.push_back(Io::Wakeup(ANNOUNCE_INTERVAL));
+            self.last_announce = now;
+        }
+        if now - self.last_prune >= PRUNE_INTERVAL {
+            debug!("Running 'prune' task...");
+
+            self.prune_routing_entries();
+            self.context.io.push_back(Io::Wakeup(PRUNE_INTERVAL));
+            self.last_prune = now;
+        }
+    }
+
+    fn command(&mut self, cmd: Self::Command) {
+        debug!("Command {:?}", cmd);
+
+        match cmd {
+            Command::Connect(addr) => self.context.connect(addr),
+        }
+    }
+
+    fn attempted(&mut self, addr: &std::net::SocketAddr) {
+        let id = addr.ip();
+        let persistent = self.context.config.is_persistent(addr);
+        let mut peer = self
+            .peers
+            .entry(id)
+            .or_insert_with(|| Peer::new(*addr, Link::Outbound, persistent));
+
+        peer.attempts += 1;
+    }
+
+    fn connected(
+        &mut self,
+        addr: std::net::SocketAddr,
+        _local_addr: &std::net::SocketAddr,
+        link: Link,
+    ) {
+        let id = addr.ip();
+
+        debug!("Connected to {} ({:?})", id, link);
+
+        // For outbound connections, we are the first to say "Hello".
+        // For inbound connections, we wait for the remote to say "Hello" first.
+        // TODO: How should we deal with multiple peers connecting from the same IP address?
+        if link.is_outbound() {
+            let since = self.local_time();
+
+            if let Some(peer) = self.peers.get_mut(&id) {
+                self.context
+                    .write_all(peer.addr, [Message::hello(), Message::get_inventory([])]);
+
+                peer.state = PeerState::Negotiated { since };
+                peer.attempts = 0;
+            }
+        } else {
+            self.peers.insert(
+                id,
+                Peer::new(
+                    addr,
+                    Link::Inbound,
+                    self.context.config.is_persistent(&addr),
+                ),
+            );
+        }
+    }
+
+    fn disconnected(
+        &mut self,
+        addr: &std::net::SocketAddr,
+        reason: nakamoto::DisconnectReason<Self::DisconnectReason>,
+    ) {
+        let since = self.local_time();
+        let id = addr.ip();
+
+        debug!("Disconnected from {} ({})", id, reason);
+
+        if let Some(peer) = self.peers.get_mut(&id) {
+            peer.state = PeerState::Disconnected { since };
+
+            // Attempt to re-connect to persistent peers.
+            if self.context.config.is_persistent(addr) && peer.attempts < MAX_CONNECTION_ATTEMPTS {
+                if reason.is_dial_err() {
+                    return;
+                }
+                if let nakamoto::DisconnectReason::Protocol(r) = reason {
+                    if !r.is_transient() {
+                        return;
+                    }
+                }
+                // TODO: Eventually we want a delay before attempting a reconnection,
+                // with exponential back-off.
+                debug!("Reconnecting to {} (attempts={})...", id, peer.attempts);
+
+                // TODO: Try to reconnect only if the peer was attempted. A disconnect without
+                // even a successful attempt means that we're unlikely to be able to reconnect.
+
+                self.context.connect(*addr);
+            } else {
+                // TODO: Non-persistent peers should be removed from the
+                // map here or at some later point.
+            }
+        }
+    }
+
+    fn received_bytes(&mut self, addr: &std::net::SocketAddr, bytes: &[u8]) {
+        let peer = addr.ip();
+
+        let (peer, msgs) = if let Some(peer) = self.peers.get_mut(&peer) {
+            let decoder = &mut peer.inbox;
+            decoder.input(bytes);
+
+            let mut msgs = Vec::with_capacity(1);
+            loop {
+                match decoder.decode_next() {
+                    Ok(Some(msg)) => msgs.push(msg),
+                    Ok(None) => break,
+
+                    Err(_err) => {
+                        // TODO: Disconnect peer.
+                        return;
+                    }
+                }
+            }
+            (peer, msgs)
+        } else {
+            return;
+        };
+
+        for msg in msgs {
+            peer.received(msg, &mut self.context);
+        }
+    }
+}
+
+impl<S, T> Deref for Protocol<S, T> {
+    type Target = Context<S, T>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.context
+    }
+}
+
+impl<S, T> DerefMut for Protocol<S, T> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.context
+    }
+}
+
+#[derive(Debug, Clone)]
+pub enum DisconnectReason {
+    User,
+}
+
+impl DisconnectReason {
+    fn is_transient(&self) -> bool {
+        match self {
+            Self::User => false,
+        }
+    }
+}
+
+impl From<DisconnectReason> for nakamoto_net::DisconnectReason<DisconnectReason> {
+    fn from(reason: DisconnectReason) -> Self {
+        nakamoto_net::DisconnectReason::Protocol(reason)
+    }
+}
+
+impl fmt::Display for DisconnectReason {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::User => write!(f, "user"),
+        }
+    }
+}
+
+impl<S, T> Iterator for Protocol<S, T> {
+    type Item = Io<(), DisconnectReason>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.context.io.pop_front()
+    }
+}
+
+/// Result of a project lookup.
+#[derive(Debug)]
+pub struct Lookup {
+    /// Whether the project was found locally or not.
+    pub local: Option<Refs>,
+    /// A list of remote peers on which the project is known to exist.
+    pub remote: Vec<PeerId>,
+}
+
+/// Global protocol state used across peers.
+#[derive(Debug)]
+pub struct Context<S, T> {
+    /// Protocol configuration.
+    config: Config,
+    /// Tracks the location of projects.
+    routing: Routing,
+    /// Outgoing I/O queue.
+    io: VecDeque<Io<(), DisconnectReason>>,
+    /// Clock. Tells the time.
+    clock: RefClock,
+    /// Sequence number used for inventory messages.
+    seq: u64,
+    /// Project storage.
+    storage: T,
+    /// Peer address manager.
+    addrmgr: AddressManager<S>,
+    /// Source of entropy.
+    rng: Rng,
+}
+
+impl<S, T> Context<S, T>
+where
+    T: storage::ReadStorage,
+{
+    fn new(
+        config: Config,
+        clock: RefClock,
+        storage: T,
+        addrmgr: AddressManager<S>,
+        rng: Rng,
+    ) -> Self {
+        Self {
+            config,
+            clock,
+            routing: HashMap::with_hasher(rng.clone().into()),
+            io: VecDeque::new(),
+            seq: 0,
+            storage,
+            addrmgr,
+            rng,
+        }
+    }
+
+    /// Process a peer inventory announcement by updating our routing table.
+    fn process_inventory(&mut self, from: PeerId, inventory: Inventory) {
+        for (proj_id, _refs) in inventory {
+            let inventory = self
+                .routing
+                .entry(proj_id)
+                .or_insert_with(|| HashSet::with_hasher(self.rng.clone().into()));
+
+            // TODO: If we're tracking this project, check the refs to see if we need to
+            // fetch updates from this peer.
+
+            inventory.insert(from);
+        }
+    }
+}
+
+impl<S, T> Context<S, T> {
+    /// Connect to a peer.
+    fn connect(&mut self, addr: net::SocketAddr) {
+        // TODO: Make sure we don't try to connect more than once to the same address.
+        self.io.push_back(Io::Connect(addr));
+    }
+
+    fn write_all(&mut self, remote: net::SocketAddr, msgs: impl IntoIterator<Item = Message>) {
+        let mut buf = io::Cursor::new(Vec::new());
+
+        for msg in msgs {
+            debug!("Write {:?} to {}", &msg, remote.ip());
+
+            serde_json::to_writer(&mut buf, &Envelope::from(msg)).unwrap();
+        }
+        self.io.push_back(Io::Write(remote, buf.into_inner()));
+    }
+
+    fn write(&mut self, remote: net::SocketAddr, msg: Message) {
+        debug!("Write {:?} to {}", &msg, remote.ip());
+
+        let bytes = serde_json::to_vec(&Envelope::from(msg)).unwrap();
+
+        self.io.push_back(Io::Write(remote, bytes));
+    }
+}
+
+#[derive(Debug)]
+pub struct Peers(AddressBook<PeerId, Peer>);
+
+impl Peers {
+    pub fn new(rng: Rng) -> Self {
+        Self(AddressBook::new(rng))
+    }
+
+    /// Iterator over fully negotiated peers.
+    pub fn negotiated(&self) -> impl Iterator<Item = (&IpAddr, &Peer)> + Clone {
+        self.0.iter().filter(move |(_, p)| p.is_negotiated())
+    }
+}
+
+impl Deref for Peers {
+    type Target = AddressBook<PeerId, Peer>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl DerefMut for Peers {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+#[derive(Debug, Default)]
+enum PeerState {
+    /// Initial peer state. For outgoing peers this
+    /// means we've attempted a connection. For incoming
+    /// peers, this means they've successfully connected
+    /// to us.
+    #[default]
+    Initial,
+    /// State after successful handshake.
+    Negotiated { since: LocalTime },
+    /// When a peer is disconnected.
+    Disconnected { since: LocalTime },
+}
+
+#[derive(Debug)]
+pub struct Peer {
+    /// Peer address.
+    addr: net::SocketAddr,
+    /// Inbox for incoming messages from peer.
+    inbox: Decoder,
+    /// Peer connection state.
+    state: PeerState,
+    /// Connection direction.
+    link: Link,
+    /// Whether we should attempt to re-connect
+    /// to this peer upon disconnection.
+    persistent: bool,
+    /// Connection attempts. For persistent peers, Tracks
+    /// how many times we've attempted to connect. We reset this to zero
+    /// upon successful connection.
+    attempts: usize,
+}
+
+impl Peer {
+    fn new(addr: net::SocketAddr, link: Link, persistent: bool) -> Self {
+        Self {
+            addr,
+            inbox: Decoder::new(256),
+            state: PeerState::default(),
+            link,
+            persistent,
+            attempts: 0,
+        }
+    }
+
+    fn id(&self) -> PeerId {
+        self.addr.ip()
+    }
+
+    fn is_negotiated(&self) -> bool {
+        matches!(self.state, PeerState::Negotiated { .. })
+    }
+
+    fn received<S, T>(&mut self, envelope: Envelope, ctx: &mut Context<S, T>)
+    where
+        T: storage::ReadStorage,
+    {
+        if envelope.magic != NETWORK_MAGIC {
+            // TODO: Disconnect
+            return;
+        }
+        debug!("Received {:?} from {}", &envelope.msg, self.id());
+
+        match envelope.msg {
+            Message::Hello { .. } => {
+                if let PeerState::Initial = self.state {
+                    // TODO: Check version.
+                    // Nb. This is a very primitive handshake. Eventually we should have anyhow
+                    // extra "acknowledgment" message sent when the `Hello` is well received.
+                    if self.link.is_inbound() {
+                        ctx.write_all(self.addr, [Message::hello(), Message::get_inventory([])]);
+                    }
+                    self.state = PeerState::Negotiated {
+                        since: ctx.clock.local_time(),
+                    };
+                } else {
+                    // TODO: Handle misbehavior.
+                }
+            }
+            Message::GetInventory { .. } => {
+                // TODO: Handle partial inventory requests.
+                let inventory = Message::inventory(ctx).unwrap();
+                ctx.write(self.addr, inventory);
+            }
+            Message::Inventory { inv, .. } => {
+                ctx.process_inventory(self.id(), inv);
+            }
+            Message::GetAddrs => {
+                // TODO: Send peer addresses.
+                todo!();
+            }
+            Message::Addrs { .. } => {
+                // TODO: Update address book.
+                todo!();
+            }
+        }
+    }
+}

node/src/storage.rs

@@ -0,0 +1,194 @@
+use std::marker::PhantomData;
+use std::ops::{Deref, DerefMut};
+use std::path::Path;
+use std::{fmt, fs, io, net};
+
+use git_ref_format::refspec;
+use once_cell::sync::Lazy;
+use radicle_git_ext as git_ext;
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+
+pub use radicle_git_ext::Oid;
+
+use crate::collections::HashMap;
+use crate::identity;
+use crate::identity::{IdError, ProjId, UserId};
+
+pub static RAD_ID_GLOB: Lazy<refspec::PatternString> =
+    Lazy::new(|| refspec::pattern!("refs/namespaces/*/refs/rad/id"));
+pub static IDENTITY_PATH: Lazy<&Path> = Lazy::new(|| Path::new(".rad/identity.toml"));
+
+pub type BranchName = String;
+pub type Inventory = Vec<(ProjId, Refs)>;
+
+/// Storage error.
+#[derive(Error, Debug)]
+pub enum Error {
+    #[error("invalid git reference")]
+    InvalidRef,
+    #[error("git: {0}")]
+    Git(#[from] git2::Error),
+    #[error("id: {0}")]
+    ProjId(#[from] IdError),
+    #[error("i/o: {0}")]
+    Io(#[from] io::Error),
+    #[error("doc: {0}")]
+    Doc(#[from] identity::DocError),
+    #[error("invalid repository head")]
+    InvalidHead,
+}
+
+#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, Eq)]
+pub struct Refs {
+    heads: HashMap<BranchName, Oid>,
+}
+
+impl From<HashMap<BranchName, Oid>> for Refs {
+    fn from(heads: HashMap<BranchName, Oid>) -> Self {
+        Self { heads }
+    }
+}
+
+pub type RemoteId = UserId;
+pub type RefName = String;
+
+/// Verified (used as type witness).
+pub struct Verified;
+/// Unverified (used as type witness).
+pub struct Unverified;
+
+/// Project remotes. Tracks the git state of a project.
+#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, Eq)]
+pub struct Remotes<V>(HashMap<RemoteId, Remote<V>>);
+
+/// A project remote.
+#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, Eq)]
+pub struct Remote<V> {
+    /// Git references published under this remote, and their hashes.
+    refs: HashMap<RefName, Oid>,
+    /// Whether this remote is of a project delegate.
+    delegate: bool,
+    /// Whether the remote is verified or not, ie. whether its signed refs were checked.
+    verified: PhantomData<V>,
+}
+
+pub trait ReadStorage {
+    fn get(&self, proj: &ProjId) -> Result<Option<Refs>, Error>;
+    fn inventory(&self) -> Result<Inventory, Error>;
+}
+
+pub trait WriteStorage {
+    /// Fetch a project from a remote peer.
+    fn fetch(&mut self, proj: &ProjId, remote: &net::SocketAddr) -> Result<(), Error>;
+}
+
+impl<T, S> ReadStorage for T
+where
+    T: Deref<Target = S>,
+    S: ReadStorage,
+{
+    fn inventory(&self) -> Result<Inventory, Error> {
+        self.deref().inventory()
+    }
+
+    fn get(&self, proj: &ProjId) -> Result<Option<Refs>, Error> {
+        self.deref().get(proj)
+    }
+}
+
+impl<T, S> WriteStorage for T
+where
+    T: DerefMut<Target = S>,
+    S: WriteStorage,
+{
+    fn fetch(&mut self, proj: &ProjId, remote: &net::SocketAddr) -> Result<(), Error> {
+        self.deref_mut().fetch(proj, remote)
+    }
+}
+
+pub struct Storage {
+    backend: git2::Repository,
+}
+
+impl fmt::Debug for Storage {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "Storage(..)")
+    }
+}
+
+impl ReadStorage for Storage {
+    fn get(&self, _id: &ProjId) -> Result<Option<Refs>, Error> {
+        todo!()
+    }
+
+    fn inventory(&self) -> Result<Inventory, Error> {
+        let glob: String = RAD_ID_GLOB.clone().into();
+        let refs = self.backend.references_glob(glob.as_str())?;
+        let mut projs = Vec::new();
+
+        for r in refs {
+            let r = r?;
+            let name = r.name().ok_or(Error::InvalidRef)?;
+            let id = ProjId::from_ref(name)?;
+
+            projs.push((id, Refs::default()));
+        }
+        Ok(projs)
+    }
+}
+
+impl WriteStorage for Storage {
+    fn fetch(&mut self, _id: &ProjId, _remote: &net::SocketAddr) -> Result<(), Error> {
+        todo!()
+    }
+}
+
+impl Storage {
+    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self, git2::Error> {
+        let path = path.as_ref();
+        let backend = match git2::Repository::open_bare(path) {
+            Err(e) if git_ext::is_not_found_err(&e) => {
+                let backend = git2::Repository::init_opts(
+                    path,
+                    git2::RepositoryInitOptions::new()
+                        .bare(true)
+                        .no_reinit(true)
+                        .external_template(false),
+                )?;
+
+                Ok(backend)
+            }
+            Ok(repo) => Ok(repo),
+            Err(e) => Err(e),
+        }?;
+
+        Ok(Self { backend })
+    }
+
+    pub fn create(
+        &self,
+        repo: &git2::Repository,
+        identity: impl Into<identity::Doc>,
+    ) -> Result<(ProjId, git2::Reference), Error> {
+        let doc = identity.into();
+        let file = fs::OpenOptions::new()
+            .create_new(true)
+            .write(true)
+            .open(*IDENTITY_PATH)?;
+        let id = doc.write(file)?;
+        let ref_name = RAD_ID_GLOB.replace('*', &id.encode());
+        let oid = repo.head()?.target().ok_or(Error::InvalidHead)?;
+        let reference = self.backend.reference(&ref_name, oid, false, "")?;
+
+        // TODO: Push project to monorepo.
+
+        Ok((id, reference))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn test_storage() {}
+}

node/src/test.rs

@@ -0,0 +1,6 @@
+mod arbitrary;
+mod assert;
+mod logger;
+mod peer;
+mod storage;
+mod tests;

node/src/test/arbitrary.rs

@@ -0,0 +1,58 @@
+use crate::collections::HashMap;
+use crate::hash;
+use crate::identity::{ProjId, UserId};
+use crate::storage;
+
+impl quickcheck::Arbitrary for storage::Refs {
+    fn arbitrary(g: &mut quickcheck::Gen) -> Self {
+        let rng = fastrand::Rng::with_seed(u64::arbitrary(g));
+        let mut refs: HashMap<storage::BranchName, storage::Oid> = HashMap::with_hasher(rng.into());
+        let mut bytes: [u8; 20] = [0; 20];
+        let names = &["master", "dev", "feature/1", "feature/2", "feature/3"];
+
+        for _ in 0..g.size().min(2) {
+            if let Some(name) = g.choose(names) {
+                for byte in &mut bytes {
+                    *byte = u8::arbitrary(g);
+                }
+                let oid = storage::Oid::try_from(&bytes[..]).unwrap();
+                refs.insert(name.to_string(), oid);
+            }
+        }
+        storage::Refs::from(refs)
+    }
+}
+
+impl quickcheck::Arbitrary for ProjId {
+    fn arbitrary(g: &mut quickcheck::Gen) -> Self {
+        let digest = hash::Digest::arbitrary(g);
+        ProjId::from(digest)
+    }
+}
+
+impl quickcheck::Arbitrary for hash::Digest {
+    fn arbitrary(g: &mut quickcheck::Gen) -> Self {
+        let mut bytes: [u8; 32] = [0; 32];
+
+        for byte in &mut bytes {
+            *byte = u8::arbitrary(g);
+        }
+        hash::Digest::from(bytes)
+    }
+}
+
+impl quickcheck::Arbitrary for UserId {
+    fn arbitrary(g: &mut quickcheck::Gen) -> Self {
+        use ed25519_consensus::SigningKey;
+
+        let mut bytes: [u8; 32] = [0; 32];
+
+        for byte in &mut bytes {
+            *byte = u8::arbitrary(g);
+        }
+        let sk = SigningKey::from(bytes);
+        let vk = sk.verification_key();
+
+        UserId(vk)
+    }
+}

node/src/test/assert.rs

@@ -0,0 +1,296 @@
+// Copyright (c) 2016 Murarth
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of this software
+// and associated documentation files (the "Software"), to deal in the Software without
+// restriction, including without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all copies or
+// substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING
+// BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+// NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+//! Provides a macro, `assert_matches!`, which tests whether a value
+//! matches a given pattern, causing a panic if the match fails.
+//!
+//! See the macro [`assert_matches!`] documentation for more information.
+//!
+//! Also provides a debug-only counterpart, [`debug_assert_matches!`].
+//!
+//! See the macro [`debug_assert_matches!`] documentation for more information
+//! about this macro.
+//!
+//! [`assert_matches!`]: macro.assert_matches.html
+//! [`debug_assert_matches!`]: macro.debug_assert_matches.html
+
+#![deny(missing_docs)]
+
+/// Asserts that an expression matches a given pattern.
+///
+/// A guard expression may be supplied to add further restrictions to the
+/// expected value of the expression.
+///
+/// A `match` arm may be supplied to perform additional assertions or to yield
+/// a value from the macro invocation.
+///
+#[macro_export]
+macro_rules! assert_matches {
+    ( $e:expr , $($pat:pat_param)|+ ) => {
+        match $e {
+            $($pat)|+ => (),
+            ref e => panic!("assertion failed: `{:?}` does not match `{}`",
+                e, stringify!($($pat)|+))
+        }
+    };
+    ( $e:expr , $($pat:pat_param)|+ if $cond:expr ) => {
+        match $e {
+            $($pat)|+ if $cond => (),
+            ref e => panic!("assertion failed: `{:?}` does not match `{}`",
+                e, stringify!($($pat)|+ if $cond))
+        }
+    };
+    ( $e:expr , $($pat:pat_param)|+ => $arm:expr ) => {
+        match $e {
+            $($pat)|+ => $arm,
+            ref e => panic!("assertion failed: `{:?}` does not match `{}`",
+                e, stringify!($($pat)|+))
+        }
+    };
+    ( $e:expr , $($pat:pat_param)|+ if $cond:expr => $arm:expr ) => {
+        match $e {
+            $($pat)|+ if $cond => $arm,
+            ref e => panic!("assertion failed: `{:?}` does not match `{}`",
+                e, stringify!($($pat)|+ if $cond))
+        }
+    };
+    ( $e:expr , $($pat:pat_param)|+ , $($arg:tt)* ) => {
+        match $e {
+            $($pat)|+ => (),
+            ref e => panic!("assertion failed: `{:?}` does not match `{}`: {}",
+                e, stringify!($($pat)|+), format_args!($($arg)*))
+        }
+    };
+    ( $e:expr , $($pat:pat_param)|+ if $cond:expr , $($arg:tt)* ) => {
+        match $e {
+            $($pat)|+ if $cond => (),
+            ref e => panic!("assertion failed: `{:?}` does not match `{}`: {}",
+                e, stringify!($($pat)|+ if $cond), format_args!($($arg)*))
+        }
+    };
+    ( $e:expr , $($pat:pat_param)|+ => $arm:expr , $($arg:tt)* ) => {
+        match $e {
+            $($pat)|+ => $arm,
+            ref e => panic!("assertion failed: `{:?}` does not match `{}`: {}",
+                e, stringify!($($pat)|+), format_args!($($arg)*))
+        }
+    };
+    ( $e:expr , $($pat:pat_param)|+ if $cond:expr => $arm:expr , $($arg:tt)* ) => {
+        match $e {
+            $($pat)|+ if $cond => $arm,
+            ref e => panic!("assertion failed: `{:?}` does not match `{}`: {}",
+                e, stringify!($($pat)|+ if $cond), format_args!($($arg)*))
+        }
+    };
+}
+
+/// Asserts that an expression matches a given pattern.
+///
+/// Unlike [`assert_matches!`], `debug_assert_matches!` statements are only enabled
+/// in non-optimized builds by default. An optimized build will omit all
+/// `debug_assert_matches!` statements unless `-C debug-assertions` is passed
+/// to the compiler.
+///
+/// See the macro [`assert_matches!`] documentation for more information.
+///
+/// [`assert_matches!`]: macro.assert_matches.html
+#[macro_export(local_inner_macros)]
+macro_rules! debug_assert_matches {
+    ( $($tt:tt)* ) => { {
+        if _assert_matches_cfg!(debug_assertions) {
+            assert_matches!($($tt)*);
+        }
+    } }
+}
+
+#[doc(hidden)]
+#[macro_export]
+macro_rules! _assert_matches_cfg {
+    ( $($tt:tt)* ) => { cfg!($($tt)*) }
+}
+
+#[cfg(test)]
+mod test {
+    use std::panic::{catch_unwind, UnwindSafe};
+
+    #[derive(Debug)]
+    enum Foo {
+        A(i32),
+        B(&'static str),
+        C(&'static str),
+    }
+
+    #[test]
+    fn test_assert_succeed() {
+        let a = Foo::A(123);
+
+        assert_matches!(a, Foo::A(_));
+        assert_matches!(a, Foo::A(123));
+        assert_matches!(a, Foo::A(i) if i == 123);
+        assert_matches!(a, Foo::A(42) | Foo::A(123));
+
+        let b = Foo::B("foo");
+
+        assert_matches!(b, Foo::B(_));
+        assert_matches!(b, Foo::B("foo"));
+        assert_matches!(b, Foo::B(s) if s == "foo");
+        assert_matches!(b, Foo::B(s) => assert_eq!(s, "foo"));
+        assert_matches!(b, Foo::B(s) => { assert_eq!(s, "foo") });
+        assert_matches!(b, Foo::B(s) if s == "foo" => assert_eq!(s, "foo"));
+        assert_matches!(b, Foo::B(s) if s == "foo" => { assert_eq!(s, "foo") });
+
+        let c = Foo::C("foo");
+
+        assert_matches!(c, Foo::B(_) | Foo::C(_));
+        assert_matches!(c, Foo::B("foo") | Foo::C("foo"));
+        assert_matches!(c, Foo::B(s) | Foo::C(s) if s == "foo");
+        assert_matches!(c, Foo::B(s) | Foo::C(s) => assert_eq!(s, "foo"));
+        assert_matches!(c, Foo::B(s) | Foo::C(s) => { assert_eq!(s, "foo") });
+        assert_matches!(c, Foo::B(s) | Foo::C(s) if s == "foo" => assert_eq!(s, "foo"));
+        assert_matches!(c, Foo::B(s) | Foo::C(s) if s == "foo" => { assert_eq!(s, "foo") });
+    }
+
+    #[test]
+    #[should_panic]
+    fn test_assert_panic_0() {
+        let a = Foo::A(123);
+
+        assert_matches!(a, Foo::B(_));
+    }
+
+    #[test]
+    #[should_panic]
+    fn test_assert_panic_1() {
+        let b = Foo::B("foo");
+
+        assert_matches!(b, Foo::B("bar"));
+    }
+
+    #[test]
+    #[should_panic]
+    fn test_assert_panic_2() {
+        let b = Foo::B("foo");
+
+        assert_matches!(b, Foo::B(s) if s == "bar");
+    }
+
+    #[test]
+    fn test_assert_no_move() {
+        let b = &mut Foo::A(0);
+        assert_matches!(*b, Foo::A(0));
+    }
+
+    #[test]
+    fn assert_with_message() {
+        let a = Foo::A(0);
+
+        assert_matches!(a, Foo::A(_), "o noes");
+        assert_matches!(a, Foo::A(n) if n == 0, "o noes");
+        assert_matches!(a, Foo::A(n) => assert_eq!(n, 0), "o noes");
+        assert_matches!(a, Foo::A(n) => { assert_eq!(n, 0); assert!(n < 1) }, "o noes");
+        assert_matches!(a, Foo::A(n) if n == 0 => assert_eq!(n, 0), "o noes");
+        assert_matches!(a, Foo::A(n) if n == 0 => { assert_eq!(n, 0); assert!(n < 1) }, "o noes");
+        assert_matches!(a, Foo::A(_), "o noes {:?}", a);
+        assert_matches!(a, Foo::A(n) if n == 0, "o noes {:?}", a);
+        assert_matches!(a, Foo::A(n) => assert_eq!(n, 0), "o noes {:?}", a);
+        assert_matches!(a, Foo::A(n) => { assert_eq!(n, 0); assert!(n < 1) }, "o noes {:?}", a);
+        assert_matches!(a, Foo::A(_), "o noes {value:?}", value = a);
+        assert_matches!(a, Foo::A(n) if n == 0, "o noes {value:?}", value=a);
+        assert_matches!(a, Foo::A(n) => assert_eq!(n, 0), "o noes {value:?}", value=a);
+        assert_matches!(a, Foo::A(n) => { assert_eq!(n, 0); assert!(n < 1) }, "o noes {value:?}", value=a);
+        assert_matches!(a, Foo::A(n) if n == 0 => assert_eq!(n, 0), "o noes {value:?}", value=a);
+    }
+
+    fn panic_message<F>(f: F) -> String
+    where
+        F: FnOnce() + UnwindSafe,
+    {
+        let err = catch_unwind(f).expect_err("function did not panic");
+
+        *err.downcast::<String>()
+            .expect("function panicked with non-String value")
+    }
+
+    #[test]
+    fn test_panic_message() {
+        let a = Foo::A(1);
+
+        // expr, pat
+        assert_eq!(
+            panic_message(|| {
+                assert_matches!(a, Foo::B(_));
+            }),
+            r#"assertion failed: `A(1)` does not match `Foo::B(_)`"#
+        );
+
+        // expr, pat if cond
+        assert_eq!(
+            panic_message(|| {
+                assert_matches!(a, Foo::B(s) if s == "foo");
+            }),
+            r#"assertion failed: `A(1)` does not match `Foo::B(s) if s == "foo"`"#
+        );
+
+        // expr, pat => arm
+        assert_eq!(
+            panic_message(|| {
+                assert_matches!(a, Foo::B(_) => {});
+            }),
+            r#"assertion failed: `A(1)` does not match `Foo::B(_)`"#
+        );
+
+        // expr, pat if cond => arm
+        assert_eq!(
+            panic_message(|| {
+                assert_matches!(a, Foo::B(s) if s == "foo" => {});
+            }),
+            r#"assertion failed: `A(1)` does not match `Foo::B(s) if s == "foo"`"#
+        );
+
+        // expr, pat, args
+        assert_eq!(
+            panic_message(|| {
+                assert_matches!(a, Foo::B(_), "msg");
+            }),
+            r#"assertion failed: `A(1)` does not match `Foo::B(_)`: msg"#
+        );
+
+        // expr, pat if cond, args
+        assert_eq!(
+            panic_message(|| {
+                assert_matches!(a, Foo::B(s) if s == "foo", "msg");
+            }),
+            r#"assertion failed: `A(1)` does not match `Foo::B(s) if s == "foo"`: msg"#
+        );
+
+        // expr, pat => arm, args
+        assert_eq!(
+            panic_message(|| {
+                assert_matches!(a, Foo::B(_) => {}, "msg");
+            }),
+            r#"assertion failed: `A(1)` does not match `Foo::B(_)`: msg"#
+        );
+
+        // expr, pat if cond => arm, args
+        assert_eq!(
+            panic_message(|| {
+                assert_matches!(a, Foo::B(s) if s == "foo" => {}, "msg");
+            }),
+            r#"assertion failed: `A(1)` does not match `Foo::B(s) if s == "foo"`: msg"#
+        );
+    }
+}

node/src/test/logger.rs

@@ -0,0 +1,43 @@
+use log::*;
+
+struct Logger {
+    level: Level,
+}
+
+impl Log for Logger {
+    fn enabled(&self, metadata: &Metadata) -> bool {
+        metadata.level() <= self.level
+    }
+
+    fn log(&self, record: &Record) {
+        use colored::Colorize;
+
+        match record.target() {
+            "test" => {
+                println!(
+                    "{} {}",
+                    "test:".yellow(),
+                    record.args().to_string().yellow()
+                )
+            }
+            "sim" => {
+                println!("{}  {}", "sim:".bold(), record.args().to_string().bold())
+            }
+            target => {
+                if self.enabled(record.metadata()) {
+                    let s = format!("{:<8} {}", format!("{}:", target), record.args());
+                    println!("{}", s.dimmed());
+                }
+            }
+        }
+    }
+
+    fn flush(&self) {}
+}
+
+pub fn init(level: Level) {
+    let logger = Logger { level };
+
+    log::set_boxed_logger(Box::new(logger)).ok();
+    log::set_max_level(level.to_level_filter());
+}

node/src/test/peer.rs

@@ -0,0 +1,181 @@
+use std::net;
+use std::ops::{Deref, DerefMut};
+
+use log::*;
+use nakamoto_net::simulator;
+use nakamoto_net::Protocol as _;
+
+use crate::address_book::{KnownAddress, Source};
+use crate::clock::RefClock;
+use crate::collections::HashMap;
+use crate::decoder::Decoder;
+use crate::protocol::*;
+use crate::storage::{ReadStorage, WriteStorage};
+use crate::*;
+
+/// Protocol instantiation used for testing.
+pub type Protocol<S> = crate::protocol::Protocol<HashMap<net::IpAddr, KnownAddress>, S>;
+
+#[derive(Debug)]
+pub struct Peer<S> {
+    pub name: &'static str,
+    pub protocol: Protocol<S>,
+    pub ip: net::IpAddr,
+    pub rng: fastrand::Rng,
+    pub local_time: LocalTime,
+    pub local_addr: net::SocketAddr,
+
+    initialized: bool,
+}
+
+impl<S> simulator::Peer<Protocol<S>> for Peer<S>
+where
+    S: ReadStorage + WriteStorage + 'static,
+{
+    fn init(&mut self) {
+        self.initialize()
+    }
+
+    fn addr(&self) -> net::SocketAddr {
+        net::SocketAddr::new(self.ip, DEFAULT_PORT)
+    }
+}
+
+impl<S> Deref for Peer<S> {
+    type Target = Protocol<S>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.protocol
+    }
+}
+
+impl<S> DerefMut for Peer<S> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.protocol
+    }
+}
+
+impl<S> Peer<S>
+where
+    S: ReadStorage + WriteStorage,
+{
+    pub fn new(name: &'static str, ip: impl Into<net::IpAddr>, storage: S) -> Self {
+        Self::config(
+            name,
+            Config::default(),
+            ip,
+            vec![],
+            storage,
+            fastrand::Rng::new(),
+        )
+    }
+
+    pub fn config(
+        name: &'static str,
+        config: Config,
+        ip: impl Into<net::IpAddr>,
+        addrs: Vec<(net::SocketAddr, Source)>,
+        storage: S,
+        rng: fastrand::Rng,
+    ) -> Self {
+        let addrs = addrs
+            .into_iter()
+            .map(|(addr, src)| (addr.ip(), KnownAddress::new(addr, src, None)))
+            .collect();
+        let local_time = LocalTime::now();
+        let clock = RefClock::from(local_time);
+        let protocol = Protocol::new(config, clock, storage, addrs, rng.clone());
+        let ip = ip.into();
+        let local_addr = net::SocketAddr::new(ip, rng.u16(..));
+
+        Self {
+            name,
+            protocol,
+            ip,
+            local_addr,
+            rng,
+            local_time,
+            initialized: false,
+        }
+    }
+
+    pub fn initialize(&mut self) {
+        if !self.initialized {
+            info!("{}: Initializing: address = {}", self.name, self.ip);
+
+            self.initialized = true;
+            self.protocol.initialize(LocalTime::now());
+        }
+    }
+
+    pub fn receive(&mut self, peer: &net::SocketAddr, msg: Message) {
+        let bytes = serde_json::to_vec(&Envelope {
+            magic: NETWORK_MAGIC,
+            msg,
+        })
+        .unwrap();
+
+        self.protocol.received_bytes(peer, &bytes);
+    }
+
+    pub fn connect_from(&mut self, remote: &net::SocketAddr) {
+        let local = net::SocketAddr::new(self.ip, self.rng.u16(..));
+
+        self.initialize();
+        self.protocol.connected(*remote, &local, Link::Inbound);
+        self.receive(remote, Message::hello());
+        self.receive(
+            remote,
+            Message::Inventory {
+                seq: 0,
+                inv: vec![],
+            },
+        );
+
+        let mut msgs = self.messages(remote);
+        msgs.find(|m| matches!(m, Message::Hello { .. }))
+            .expect("`hello` is sent");
+        msgs.find(|m| matches!(m, Message::GetInventory { .. }))
+            .expect("`get-inventory` is sent");
+    }
+
+    pub fn connect_to(&mut self, remote: &net::SocketAddr) {
+        self.initialize();
+        self.protocol.attempted(remote);
+        self.protocol
+            .connected(*remote, &self.local_addr, Link::Outbound);
+
+        let mut msgs = self.messages(remote);
+        msgs.find(|m| matches!(m, Message::Hello { .. }))
+            .expect("`hello` is sent");
+        msgs.find(|m| matches!(m, Message::GetInventory { .. }))
+            .expect("`get-inventory` is sent");
+
+        self.receive(remote, Message::hello());
+    }
+
+    /// Get outgoing messages sent from this peer to the remote address.
+    pub fn messages(&mut self, remote: &net::SocketAddr) -> impl Iterator<Item = Message> {
+        let mut stream = Decoder::<Envelope>::new(2048);
+        let mut msgs = Vec::new();
+
+        for o in self.protocol.outbox().iter() {
+            match o {
+                Io::Write(a, bytes) if a == remote => {
+                    stream.input(bytes);
+                }
+                _ => {}
+            }
+        }
+
+        while let Some(envelope) = stream.decode_next().unwrap() {
+            msgs.push(envelope.msg);
+        }
+        msgs.into_iter()
+    }
+
+    /// Get a draining iterator over the peers's I/O outbox.
+    pub fn outbox(&mut self) -> impl Iterator<Item = Io<(), DisconnectReason>> + '_ {
+        self.protocol.outbox().drain(..)
+    }
+}

node/src/test/storage.rs

@@ -0,0 +1,39 @@
+use std::net;
+
+use crate::identity::ProjId;
+use crate::storage::{Error, Inventory, ReadStorage, Refs, WriteStorage};
+
+pub struct MockStorage {
+    pub inventory: Inventory,
+}
+
+impl MockStorage {
+    pub fn new(inventory: Inventory) -> Self {
+        Self { inventory }
+    }
+
+    pub fn empty() -> Self {
+        Self {
+            inventory: Vec::new(),
+        }
+    }
+}
+
+impl ReadStorage for MockStorage {
+    fn get(&self, proj: &ProjId) -> Result<Option<Refs>, Error> {
+        if let Some((_, refs)) = self.inventory.iter().find(|(id, _)| id == proj) {
+            return Ok(Some(refs.clone()));
+        }
+        Ok(None)
+    }
+
+    fn inventory(&self) -> Result<Inventory, Error> {
+        Ok(self.inventory.clone())
+    }
+}
+
+impl WriteStorage for MockStorage {
+    fn fetch(&mut self, _proj: &ProjId, _remote: &net::SocketAddr) -> Result<(), Error> {
+        todo!()
+    }
+}

node/src/test/tests.rs

@@ -0,0 +1,224 @@
+use std::io;
+use std::sync::Arc;
+
+use nakamoto_net as nakamoto;
+use nakamoto_net::simulator;
+use nakamoto_net::simulator::{Peer as _, Simulation};
+use nakamoto_net::Protocol as _;
+use quickcheck_macros::quickcheck;
+
+use crate::collections::{HashMap, HashSet};
+use crate::protocol::*;
+use crate::storage::{Inventory, ReadStorage};
+#[allow(unused)]
+use crate::test::logger;
+use crate::test::peer::Peer;
+use crate::test::storage::MockStorage;
+use crate::*;
+
+// NOTE
+//
+// If you wish to see the logs for a running test, simply add the following line to your test:
+//
+//      logger::init(log::Level::Debug);
+//
+// You may then run the test with eg. `cargo test -- --nocapture` to always show output.
+
+#[test]
+fn test_outbound_connection() {
+    let mut alice = Peer::new("alice", [8, 8, 8, 8], MockStorage::empty());
+    let bob = Peer::new("bob", [9, 9, 9, 9], MockStorage::empty());
+    let eve = Peer::new("eve", [7, 7, 7, 7], MockStorage::empty());
+
+    alice.connect_to(&bob.addr());
+    alice.connect_to(&eve.addr());
+
+    let peers = alice
+        .protocol
+        .peers()
+        .negotiated()
+        .map(|(ip, _)| *ip)
+        .collect::<Vec<_>>();
+
+    assert!(peers.contains(&eve.ip));
+    assert!(peers.contains(&bob.ip));
+}
+
+#[test]
+fn test_inbound_connection() {
+    let mut alice = Peer::new("alice", [8, 8, 8, 8], MockStorage::empty());
+    let bob = Peer::new("bob", [9, 9, 9, 9], MockStorage::empty());
+    let eve = Peer::new("eve", [7, 7, 7, 7], MockStorage::empty());
+
+    alice.connect_from(&bob.addr());
+    alice.connect_from(&eve.addr());
+
+    let peers = alice
+        .protocol
+        .peers()
+        .negotiated()
+        .map(|(ip, _)| *ip)
+        .collect::<Vec<_>>();
+
+    assert!(peers.contains(&eve.ip));
+    assert!(peers.contains(&bob.ip));
+}
+
+#[test]
+fn test_persistent_peer_connect() {
+    let rng = fastrand::Rng::new();
+    let bob = Peer::new("bob", [8, 8, 8, 8], MockStorage::empty());
+    let eve = Peer::new("eve", [9, 9, 9, 9], MockStorage::empty());
+    let config = Config {
+        connect: vec![bob.addr(), eve.addr()],
+        ..Config::default()
+    };
+    let mut alice = Peer::config(
+        "alice",
+        config,
+        [7, 7, 7, 7],
+        vec![],
+        MockStorage::empty(),
+        rng,
+    );
+
+    alice.initialize();
+
+    let mut outbox = alice.outbox();
+    assert_matches!(outbox.next(), Some(Io::Connect(a)) if a == bob.addr());
+    assert_matches!(outbox.next(), Some(Io::Connect(a)) if a == eve.addr());
+    assert_matches!(outbox.next(), None);
+}
+
+#[test]
+fn test_persistent_peer_reconnect() {
+    let mut bob = Peer::new("bob", [8, 8, 8, 8], MockStorage::empty());
+    let mut eve = Peer::new("eve", [9, 9, 9, 9], MockStorage::empty());
+    let mut alice = Peer::config(
+        "alice",
+        Config {
+            connect: vec![bob.addr(), eve.addr()],
+            ..Config::default()
+        },
+        [7, 7, 7, 7],
+        vec![],
+        MockStorage::empty(),
+        fastrand::Rng::new(),
+    );
+
+    let mut sim = Simulation::new(
+        LocalTime::now(),
+        alice.rng.clone(),
+        simulator::Options::default(),
+    )
+    .initialize([&mut alice, &mut bob, &mut eve]);
+
+    sim.run_while([&mut alice, &mut bob, &mut eve], |s| !s.is_settled());
+
+    let ips = alice
+        .peers()
+        .negotiated()
+        .map(|(ip, _)| *ip)
+        .collect::<Vec<_>>();
+    assert!(ips.contains(&bob.ip));
+    assert!(ips.contains(&eve.ip));
+
+    // ... Negotiated ...
+    //
+    // Now let's disconnect a peer.
+
+    // A transient error such as this will cause Alice to attempt a reconnection.
+    let error = Arc::new(io::Error::from(io::ErrorKind::ConnectionReset));
+
+    // A non-transient disconnect, such as one requested by the user will not trigger
+    // a reconnection.
+    alice.disconnected(
+        &eve.addr(),
+        nakamoto::DisconnectReason::DialError(error.clone()),
+    );
+    assert_matches!(alice.outbox().next(), None);
+
+    for _ in 0..MAX_CONNECTION_ATTEMPTS {
+        alice.disconnected(
+            &bob.addr(),
+            nakamoto::DisconnectReason::ConnectionError(error.clone()),
+        );
+        assert_matches!(alice.outbox().next(), Some(Io::Connect(a)) if a == bob.addr());
+        assert_matches!(alice.outbox().next(), None);
+
+        alice.attempted(&bob.addr());
+    }
+
+    // After the max connection attempts, a disconnect doesn't trigger a reconnect.
+    alice.disconnected(
+        &bob.addr(),
+        nakamoto::DisconnectReason::ConnectionError(error),
+    );
+    assert_matches!(alice.outbox().next(), None);
+}
+
+#[quickcheck]
+fn prop_inventory_exchange_dense(alice_inv: Inventory, bob_inv: Inventory, eve_inv: Inventory) {
+    let rng = fastrand::Rng::new();
+    let alice = Peer::new("alice", [7, 7, 7, 7], MockStorage::new(alice_inv.clone()));
+    let mut bob = Peer::new("bob", [8, 8, 8, 8], MockStorage::new(bob_inv.clone()));
+    let mut eve = Peer::new("eve", [9, 9, 9, 9], MockStorage::new(eve_inv.clone()));
+    let mut routing = Routing::with_hasher(rng.clone().into());
+
+    for (inv, peer) in &[
+        (alice_inv, alice.addr().ip()),
+        (bob_inv, bob.addr().ip()),
+        (eve_inv, eve.addr().ip()),
+    ] {
+        for (proj, _) in inv {
+            routing
+                .entry(proj.clone())
+                .or_insert_with(|| HashSet::with_hasher(rng.clone().into()))
+                .insert(*peer);
+        }
+    }
+
+    // Fully-connected.
+    bob.command(Command::Connect(alice.addr()));
+    bob.command(Command::Connect(eve.addr()));
+    eve.command(Command::Connect(alice.addr()));
+    eve.command(Command::Connect(bob.addr()));
+
+    let mut peers: HashMap<_, _> = [(alice.ip, alice), (bob.ip, bob), (eve.ip, eve)]
+        .into_iter()
+        .collect();
+    let mut simulator = Simulation::new(LocalTime::now(), rng, simulator::Options::default())
+        .initialize(peers.values_mut());
+
+    simulator.run_while(peers.values_mut(), |s| !s.is_settled());
+
+    for (proj_id, remotes) in &routing {
+        for peer in peers.values() {
+            let lookup = peer.lookup(proj_id);
+
+            if lookup.local.is_some() {
+                peer.storage()
+                    .get(proj_id)
+                    .expect("There are no errors querying storage")
+                    .expect("The project is available locally");
+            } else {
+                for remote in &lookup.remote {
+                    peers[remote]
+                        .storage()
+                        .get(proj_id)
+                        .expect("There are no errors querying storage")
+                        .expect("The project is available remotely");
+                }
+                assert!(
+                    !lookup.remote.is_empty(),
+                    "There are remote locations for the project"
+                );
+                assert_eq!(
+                    &lookup.remote.into_iter().collect::<HashSet<_>>(),
+                    remotes,
+                    "The remotes match the global routing table"
+                );
+            }
+        }
+    }
+}