X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Flib.rs;h=37bbe4cb0338170998acef7ae0032c7056f7ea86;hb=refs%2Fheads%2F2023-07-further-opt;hp=25ce1a2f5015c2a23dd76479fd12f831a854903d;hpb=19ac4113f2b22dc96345dcb9737d799e4135c826;p=rapid-gossip-sync-server diff --git a/src/lib.rs b/src/lib.rs index 25ce1a2..37bbe4c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,7 +4,6 @@ #![deny(non_upper_case_globals)] #![deny(non_camel_case_types)] #![deny(non_snake_case)] -#![deny(unused_mut)] #![deny(unused_variables)] #![deny(unused_imports)] @@ -15,15 +14,13 @@ use std::fs::File; use std::io::BufReader; use std::sync::Arc; -use bitcoin::blockdata::constants::genesis_block; -use bitcoin::secp256k1::PublicKey; -use lightning::routing::gossip::NetworkGraph; +use lightning::routing::gossip::{NetworkGraph, NodeId}; use lightning::util::ser::{ReadableArgs, Writeable}; use tokio::sync::mpsc; use crate::lookup::DeltaSet; use crate::persistence::GossipPersister; -use crate::serialization::UpdateSerializationMechanism; +use crate::serialization::UpdateSerialization; use crate::snapshot::Snapshotter; use crate::types::TestLogger; @@ -38,6 +35,12 @@ mod config; mod hex_utils; mod verifier; +/// The purpose of this prefix is to identify the serialization format, should other rapid gossip +/// sync formats arise in the future. +/// +/// The fourth byte is the protocol version in case our format gets updated. +const GOSSIP_PREFIX: [u8; 4] = [76, 68, 75, 1]; + pub struct RapidSyncProcessor { network_graph: Arc>, } @@ -60,15 +63,14 @@ impl RapidSyncProcessor { let mut buffered_reader = BufReader::new(file); let network_graph_result = NetworkGraph::read(&mut buffered_reader, logger); if let Ok(network_graph) = network_graph_result { - network_graph.remove_stale_channels_and_tracking(); println!("Initialized from cached network graph!"); network_graph } else { println!("Initialization from cached network graph failed: {}", network_graph_result.err().unwrap()); - NetworkGraph::new(genesis_block(network).header.block_hash(), logger) + NetworkGraph::new(network, logger) } } else { - NetworkGraph::new(genesis_block(network).header.block_hash(), logger) + NetworkGraph::new(network, logger) }; let arc_network_graph = Arc::new(network_graph); Self { @@ -103,7 +105,38 @@ impl RapidSyncProcessor { } } -async fn serialize_delta(network_graph: Arc>, last_sync_timestamp: u32, consider_intermediate_updates: bool) -> SerializedResponse { +/// This method generates a no-op blob that can be used as a delta where none exists. +/// +/// The primary purpose of this method is the scenario of a client retrieving and processing a +/// given snapshot, and then immediately retrieving the would-be next snapshot at the timestamp +/// indicated by the one that was just processed. +/// Previously, there would not be a new snapshot to be processed for that particular timestamp yet, +/// and the server would return a 404 error. +/// +/// In principle, this method could also be used to address another unfortunately all too common +/// pitfall: requesting snapshots from intermediate timestamps, i. e. those that are not multiples +/// of our granularity constant. Note that for that purpose, this method could be very dangerous, +/// because if consumed, the `timestamp` value calculated here will overwrite the timestamp that +/// the client previously had, which could result in duplicated or omitted gossip down the line. +fn serialize_empty_blob(current_timestamp: u64) -> Vec { + let mut blob = GOSSIP_PREFIX.to_vec(); + + let network = config::network(); + let genesis_block = bitcoin::blockdata::constants::genesis_block(network); + let chain_hash = genesis_block.block_hash(); + chain_hash.write(&mut blob).unwrap(); + + let blob_timestamp = Snapshotter::round_down_to_nearest_multiple(current_timestamp, config::SNAPSHOT_CALCULATION_INTERVAL as u64) as u32; + blob_timestamp.write(&mut blob).unwrap(); + + 0u32.write(&mut blob).unwrap(); // node count + 0u32.write(&mut blob).unwrap(); // announcement count + 0u32.write(&mut blob).unwrap(); // update count + + blob +} + +async fn serialize_delta(network_graph: Arc>, last_sync_timestamp: u32) -> SerializedResponse { let (client, connection) = lookup::connect_to_db().await; network_graph.remove_stale_channels_and_tracking(); @@ -120,27 +153,26 @@ async fn serialize_delta(network_graph: Arc>, last_sync // chain hash only necessary if either channel announcements or non-incremental updates are present // for announcement-free incremental-only updates, chain hash can be skipped - let mut node_id_set: HashSet<[u8; 33]> = HashSet::new(); - let mut node_id_indices: HashMap<[u8; 33], usize> = HashMap::new(); - let mut node_ids: Vec = Vec::new(); + let mut node_id_set: HashSet = HashSet::new(); + let mut node_id_indices: HashMap = HashMap::new(); + let mut node_ids: Vec = Vec::new(); let mut duplicate_node_ids: i32 = 0; - let mut get_node_id_index = |node_id: PublicKey| { - let serialized_node_id = node_id.serialize(); - if node_id_set.insert(serialized_node_id) { + let mut get_node_id_index = |node_id: NodeId| { + if node_id_set.insert(node_id) { node_ids.push(node_id); let index = node_ids.len() - 1; - node_id_indices.insert(serialized_node_id, index); + node_id_indices.insert(node_id, index); return index; } duplicate_node_ids += 1; - node_id_indices[&serialized_node_id] + node_id_indices[&node_id] }; let mut delta_set = DeltaSet::new(); lookup::fetch_channel_announcements(&mut delta_set, network_graph, &client, last_sync_timestamp).await; println!("announcement channel count: {}", delta_set.len()); - lookup::fetch_channel_updates(&mut delta_set, &client, last_sync_timestamp, consider_intermediate_updates).await; + lookup::fetch_channel_updates(&mut delta_set, &client, last_sync_timestamp).await; println!("update-fetched channel count: {}", delta_set.len()); lookup::filter_delta_set(&mut delta_set); println!("update-filtered channel count: {}", delta_set.len()); @@ -177,11 +209,11 @@ async fn serialize_delta(network_graph: Arc>, last_sync let mut update_count_full = 0; let mut update_count_incremental = 0; for current_update in serialization_details.updates { - match ¤t_update.mechanism { - UpdateSerializationMechanism::Full => { + match ¤t_update { + UpdateSerialization::Full(_) => { update_count_full += 1; } - UpdateSerializationMechanism::Incremental(_) => { + UpdateSerialization::Incremental(_, _) | UpdateSerialization::Reminder(_, _) => { update_count_incremental += 1; } }; @@ -189,13 +221,13 @@ async fn serialize_delta(network_graph: Arc>, last_sync let mut stripped_update = serialization::serialize_stripped_channel_update(¤t_update, &default_update_values, previous_update_scid); output.append(&mut stripped_update); - previous_update_scid = current_update.update.short_channel_id; + previous_update_scid = current_update.scid(); } // some stats let message_count = announcement_count + update_count; - let mut prefixed_output = vec![76, 68, 75, 1]; + let mut prefixed_output = GOSSIP_PREFIX.to_vec(); // always write the chain hash serialization_details.chain_hash.write(&mut prefixed_output).unwrap();