Introduce graph sync crate for fast-forwarding through gossip data downloaded from a server.

Author: arik-so

Cargo.toml

@@ -7,6 +7,7 @@ members = [
     "lightning-net-tokio",
     "lightning-persister",
     "lightning-background-processor",
+    "lightning-graph-sync"
 ]
 
 # Our tests do actual crypo and lots of work, the tradeoff for -O1 is well worth it.

lightning-block-sync/src/http.rs

@@ -97,7 +97,7 @@ impl<'a> std::net::ToSocketAddrs for &'a HttpEndpoint {
 }
 
 /// Client for making HTTP requests.
-pub(crate) struct HttpClient {
+pub struct HttpClient {
 	address: SocketAddr,
 	stream: TcpStream,
 }

lightning-graph-sync/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "lightning-graph-sync"
+version = "0.0.101"
+authors = ["Arik Sosman <[email protected]>"]
+license = "MIT OR Apache-2.0"
+repository = "http://github.com/rust-bitcoin/rust-lightning"
+edition = "2018"
+description = """
+Utility to fetch gossip routing data from LNSync or LNSync-like server
+"""
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+lightning = { version = "0.0.101", path = "../lightning" }
+lightning-block-sync = { version = "0.0.101", path = "../lightning-block-sync", features = ["rest-client"] }
+tokio = { version = "1.0", features = [ "io-util", "net", "time" ], optional = true }
+
+[dev-dependencies]
+bitcoin = { version = "0.27", default-features = false, features = ["secp-recovery"] }
+tokio = { version = "1.0", features = [ "macros", "rt" ] }

lightning-graph-sync/src/error.rs

@@ -0,0 +1,26 @@
+use lightning::ln::msgs::{DecodeError, LightningError};
+
+pub enum GraphSyncError{
+    IOError(std::io::Error),
+    DecodeError(DecodeError),
+    LightningError(LightningError),
+    ProcessingError(String)
+}
+
+impl From<std::io::Error> for GraphSyncError {
+    fn from(error: std::io::Error) -> Self {
+        Self::IOError(error)
+    }
+}
+
+impl From<DecodeError> for GraphSyncError {
+    fn from(error: DecodeError) -> Self {
+        Self::DecodeError(error)
+    }
+}
+
+impl From<LightningError> for GraphSyncError {
+    fn from(error: LightningError) -> Self {
+        Self::LightningError(error)
+    }
+}

lightning-graph-sync/src/lib.rs

@@ -0,0 +1,89 @@
+use core::ops::Deref;
+
+use lightning::chain;
+use lightning::routing::network_graph;
+use lightning_block_sync::http::*;
+
+use crate::error::GraphSyncError;
+
+pub mod error;
+pub mod processing;
+
+pub async fn sync_network_graph<CS: Deref>(
+    network_graph: &network_graph::NetworkGraph,
+    sync_host: &str,
+    sync_port: u16,
+    sync_path: &str,
+    chain_source: &Option<CS>,
+) -> Result<(), GraphSyncError>
+    where
+        CS::Target: chain::Access,
+{
+    // make sure there is precisely one leading slash
+    let canonical_path = format!("/{}", sync_path.trim_start_matches("/"));
+    println!("Attempting sync from: http://{}:{}{}", sync_host, sync_port, canonical_path);
+    let http_endpoint = HttpEndpoint::for_host(sync_host.into()).with_port(sync_port);
+    let http_client_result = HttpClient::connect(&http_endpoint);
+    let mut http_client = http_client_result?;
+
+    let response_result = http_client
+        .get::<BinaryResponse>(&canonical_path, sync_host)
+        .await;
+
+    let response = response_result?;
+    let response_bytes = response.0;
+
+    processing::update_network_graph(&network_graph, &response_bytes[..], &chain_source)
+}
+
+#[cfg(test)]
+mod tests {
+    use bitcoin::blockdata::constants::genesis_block;
+    use bitcoin::Network;
+
+    use lightning::routing::network_graph::NetworkGraph;
+    use crate::error::GraphSyncError;
+
+    use crate::sync_network_graph;
+
+    #[test]
+    fn it_works() {
+        assert_eq!(2 + 2, 4);
+    }
+
+    #[tokio::test]
+    async fn test_http_request() {
+        let block_hash = genesis_block(Network::Bitcoin).block_hash();
+        let network_graph = NetworkGraph::new(block_hash);
+
+        let chain_source: Option<Box<dyn lightning::chain::Access>> = None;
+
+        let sync_result = sync_network_graph(
+            &network_graph,
+            "localhost",
+            3000,
+            "delta/500",
+            &chain_source,
+        ).await;
+
+        if !sync_result.is_ok() {
+            let error_string = match sync_result.as_ref().unwrap_err() {
+                GraphSyncError::IOError(error) => {
+                    error.to_string()
+                }
+                GraphSyncError::DecodeError(error) => {
+                    error.to_string()
+                }
+                GraphSyncError::LightningError(error) => {
+                    (&error.err).to_string()
+                }
+                GraphSyncError::ProcessingError(error) => {
+                    error.to_string()
+                }
+            };
+            println!("Error: {}", error_string);
+        }
+
+        assert!(sync_result.is_ok())
+    }
+}

lightning-graph-sync/src/processing.rs

@@ -0,0 +1,69 @@
+use core::ops::Deref;
+use std::io;
+use std::io::Read;
+use lightning::chain;
+
+use lightning::ln::msgs;
+use lightning::ln::wire::encode::Encode;
+use lightning::routing::network_graph;
+use lightning::util::ser::{BigSize, FixedLengthReader, Readable};
+
+use crate::error::GraphSyncError;
+
+pub fn update_network_graph<CS: Deref>(network_graph: &network_graph::NetworkGraph, update_data: &[u8], chain_source: &Option<CS>) -> Result<(), GraphSyncError> where CS::Target: chain::Access {
+    let total_input_length = update_data.len() as u64;
+    let mut read_cursor = io::Cursor::new(update_data);
+
+    let mut prefix = Vec::with_capacity(4);
+    prefix.resize(4, 0);
+    read_cursor.read_exact(&mut prefix)?;
+
+    let _deserialize_unsigned = match &prefix[..] {
+        &[71, 83, 80, 1] => {
+            return Err(GraphSyncError::ProcessingError("Signed deserialization not supported".to_string()));
+        }
+        &[76, 68, 75, 2] => true,
+        _ => {
+            return Err(GraphSyncError::ProcessingError("Prefix must equal 'LDK' and the byte 2".to_string()));
+        }
+    };
+
+    loop {
+        let message_length: BigSize = Readable::read(&mut read_cursor)?;
+        let mut restricted_cursor = FixedLengthReader::new(&mut read_cursor, message_length.0);
+
+        let message_type: u16 = Readable::read(&mut restricted_cursor)?;
+
+        let update_result = match message_type {
+            msgs::ChannelAnnouncement::TYPE => {
+                let channel_announcement: msgs::UnsignedChannelAnnouncement =
+                    Readable::read(&mut restricted_cursor).unwrap();
+                network_graph
+                    .update_channel_from_unsigned_announcement(&channel_announcement, &chain_source)
+            }
+            msgs::NodeAnnouncement::TYPE => {
+                let node_announcement: msgs::UnsignedNodeAnnouncement =
+                    Readable::read(&mut restricted_cursor)?;
+                network_graph.update_node_from_unsigned_announcement(&node_announcement)
+            }
+            msgs::ChannelUpdate::TYPE => {
+                let channel_update: msgs::UnsignedChannelUpdate =
+                    Readable::read(&mut restricted_cursor)?;
+                network_graph.update_channel_unsigned(&channel_update)
+            }
+            _ => {
+                let error_string = format!("Unexpected graph sync message type: {}", message_type);
+                return Err(GraphSyncError::ProcessingError(error_string));
+            }
+        };
+
+        // propagate the error
+        let _update = update_result?;
+
+        if read_cursor.position() == total_input_length {
+            break;
+        }
+    }
+
+    Ok(())
+}

lightning/src/ln/wire.rs

@@ -32,7 +32,7 @@ pub trait CustomMessageReader {
 /// variant contains a message from [`msgs`] or otherwise the message type if unknown.
 #[allow(missing_docs)]
 #[derive(Debug)]
-pub(crate) enum Message<T> where T: core::fmt::Debug + Type {
+pub enum Message<T> where T: core::fmt::Debug + Type {
 	Init(msgs::Init),
 	Error(msgs::ErrorMessage),
 	Ping(msgs::Ping),
@@ -117,7 +117,7 @@ impl<T> Message<T> where T: core::fmt::Debug + Type {
 /// # Errors
 ///
 /// Returns an error if the message payload code not be decoded as the specified type.
-pub(crate) fn read<R: io::Read, T, H: core::ops::Deref>(
+pub fn read<R: io::Read, T, H: core::ops::Deref>(
 	buffer: &mut R,
 	custom_reader: H,
 ) -> Result<Message<T>, msgs::DecodeError>
@@ -232,7 +232,8 @@ pub(crate) fn write<M: Type + Writeable, W: Writer>(message: &M, buffer: &mut W)
 	message.write(buffer)
 }
 
-mod encode {
+/// Publish message types
+pub mod encode {
 	/// Defines a constant type identifier for reading messages from the wire.
 	pub trait Encode {
 		/// The type identifying the message payload.

lightning/src/util/ser.rs

@@ -90,22 +90,25 @@ impl Writer for LengthCalculatingWriter {
 
 /// Essentially std::io::Take but a bit simpler and with a method to walk the underlying stream
 /// forward to ensure we always consume exactly the fixed length specified.
-pub(crate) struct FixedLengthReader<R: Read> {
+pub struct FixedLengthReader<R: Read> {
 	read: R,
 	bytes_read: u64,
 	total_bytes: u64,
 }
 impl<R: Read> FixedLengthReader<R> {
+	/// Create new fixed-length-reader
 	pub fn new(read: R, total_bytes: u64) -> Self {
 		Self { read, bytes_read: 0, total_bytes }
 	}
 
 	#[inline]
+	/// How many bytes remain in the FLR?
 	pub fn bytes_remain(&mut self) -> bool {
 		self.bytes_read != self.total_bytes
 	}
 
 	#[inline]
+	/// Consume remaining bytes, moving cursor to the end
 	pub fn eat_remaining(&mut self) -> Result<(), DecodeError> {
 		copy(self, &mut sink()).unwrap();
 		if self.bytes_read != self.total_bytes {
@@ -299,7 +302,7 @@ impl Readable for U48 {
 /// encoded in several different ways, which we must check for at deserialization-time. Thus, if
 /// you're looking for an example of a variable-length integer to use for your own project, move
 /// along, this is a rather poor design.
-pub(crate) struct BigSize(pub u64);
+pub struct BigSize(pub u64);
 impl Writeable for BigSize {
 	#[inline]
 	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), io::Error> {