feat: collect clock time offset from network peers

Author: jjyr

protocol/src/builder.rs

@@ -8,8 +8,8 @@ use crate::protocol_generated::ckb::protocol::{
     IndexTransactionBuilder, OutPoint as FbsOutPoint, OutPointBuilder,
     ProposalShortId as FbsProposalShortId, RelayMessage, RelayMessageBuilder, RelayPayload,
     Script as FbsScript, ScriptBuilder, SyncMessage, SyncMessageBuilder, SyncPayload,
-    Transaction as FbsTransaction, TransactionBuilder, UncleBlock as FbsUncleBlock,
-    UncleBlockBuilder, H256 as FbsH256,
+    Time as FbsTime, TimeBuilder, TimeMessage, TimeMessageBuilder, Transaction as FbsTransaction,
+    TransactionBuilder, UncleBlock as FbsUncleBlock, UncleBlockBuilder, H256 as FbsH256,
 };
 use crate::{short_transaction_id, short_transaction_id_keys};
 use ckb_core::block::Block;
@@ -293,6 +293,14 @@ impl<'a> FbsGetBlocks<'a> {
     }
 }
 
+impl<'a> FbsTime<'a> {
+    pub fn build<'b>(fbb: &mut FlatBufferBuilder<'b>, timestamp: u64) -> WIPOffset<FbsTime<'b>> {
+        let mut builder = TimeBuilder::new(fbb);
+        builder.add_timestamp(timestamp);
+        builder.finish()
+    }
+}
+
 impl<'a> SyncMessage<'a> {
     pub fn build_get_headers<'b>(
         fbb: &mut FlatBufferBuilder<'b>,
@@ -579,6 +587,19 @@ impl<'a> RelayMessage<'a> {
         builder.finish()
     }
 }
+
+impl<'a> TimeMessage<'a> {
+    pub fn build_time<'b>(
+        fbb: &mut FlatBufferBuilder<'b>,
+        timestamp: u64,
+    ) -> WIPOffset<TimeMessage<'b>> {
+        let fbs_time = FbsTime::build(fbb, timestamp);
+        let mut builder = TimeMessageBuilder::new(fbb);
+        builder.add_payload(fbs_time);
+        builder.finish()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

protocol/src/protocol.fbs

@@ -208,3 +208,12 @@ table FilteredBlock {
     transactions: [IndexTransaction];
     hashes: [H256];
 }
+
+table TimeMessage {
+    payload:        Time;
+}
+
+table Time {
+    timestamp: uint64;
+}
+

protocol/src/protocol_generated.rs

@@ -2997,6 +2997,158 @@ impl<'a: 'b, 'b> FilteredBlockBuilder<'a, 'b> {
   }
 }
 
+pub enum TimeMessageOffset {}
+#[derive(Copy, Clone, Debug, PartialEq)]
+
+pub struct TimeMessage<'a> {
+  pub _tab: flatbuffers::Table<'a>,
+}
+
+impl<'a> flatbuffers::Follow<'a> for TimeMessage<'a> {
+    type Inner = TimeMessage<'a>;
+    #[inline]
+    fn follow(buf: &'a [u8], loc: usize) -> Self::Inner {
+        Self {
+            _tab: flatbuffers::Table { buf: buf, loc: loc },
+        }
+    }
+}
+
+impl<'a> TimeMessage<'a> {
+    #[inline]
+    pub fn init_from_table(table: flatbuffers::Table<'a>) -> Self {
+        TimeMessage {
+            _tab: table,
+        }
+    }
+    #[allow(unused_mut)]
+    pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr>(
+        _fbb: &'mut_bldr mut flatbuffers::FlatBufferBuilder<'bldr>,
+        args: &'args TimeMessageArgs<'args>) -> flatbuffers::WIPOffset<TimeMessage<'bldr>> {
+      let mut builder = TimeMessageBuilder::new(_fbb);
+      if let Some(x) = args.payload { builder.add_payload(x); }
+      builder.finish()
+    }
+
+    pub const VT_PAYLOAD: flatbuffers::VOffsetT = 4;
+
+  #[inline]
+  pub fn payload(&self) -> Option<Time<'a>> {
+    self._tab.get::<flatbuffers::ForwardsUOffset<Time<'a>>>(TimeMessage::VT_PAYLOAD, None)
+  }
+}
+
+pub struct TimeMessageArgs<'a> {
+    pub payload: Option<flatbuffers::WIPOffset<Time<'a >>>,
+}
+impl<'a> Default for TimeMessageArgs<'a> {
+    #[inline]
+    fn default() -> Self {
+        TimeMessageArgs {
+            payload: None,
+        }
+    }
+}
+pub struct TimeMessageBuilder<'a: 'b, 'b> {
+  fbb_: &'b mut flatbuffers::FlatBufferBuilder<'a>,
+  start_: flatbuffers::WIPOffset<flatbuffers::TableUnfinishedWIPOffset>,
+}
+impl<'a: 'b, 'b> TimeMessageBuilder<'a, 'b> {
+  #[inline]
+  pub fn add_payload(&mut self, payload: flatbuffers::WIPOffset<Time<'b >>) {
+    self.fbb_.push_slot_always::<flatbuffers::WIPOffset<Time>>(TimeMessage::VT_PAYLOAD, payload);
+  }
+  #[inline]
+  pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a>) -> TimeMessageBuilder<'a, 'b> {
+    let start = _fbb.start_table();
+    TimeMessageBuilder {
+      fbb_: _fbb,
+      start_: start,
+    }
+  }
+  #[inline]
+  pub fn finish(self) -> flatbuffers::WIPOffset<TimeMessage<'a>> {
+    let o = self.fbb_.end_table(self.start_);
+    flatbuffers::WIPOffset::new(o.value())
+  }
+}
+
+pub enum TimeOffset {}
+#[derive(Copy, Clone, Debug, PartialEq)]
+
+pub struct Time<'a> {
+  pub _tab: flatbuffers::Table<'a>,
+}
+
+impl<'a> flatbuffers::Follow<'a> for Time<'a> {
+    type Inner = Time<'a>;
+    #[inline]
+    fn follow(buf: &'a [u8], loc: usize) -> Self::Inner {
+        Self {
+            _tab: flatbuffers::Table { buf: buf, loc: loc },
+        }
+    }
+}
+
+impl<'a> Time<'a> {
+    #[inline]
+    pub fn init_from_table(table: flatbuffers::Table<'a>) -> Self {
+        Time {
+            _tab: table,
+        }
+    }
+    #[allow(unused_mut)]
+    pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr>(
+        _fbb: &'mut_bldr mut flatbuffers::FlatBufferBuilder<'bldr>,
+        args: &'args TimeArgs) -> flatbuffers::WIPOffset<Time<'bldr>> {
+      let mut builder = TimeBuilder::new(_fbb);
+      builder.add_timestamp(args.timestamp);
+      builder.finish()
+    }
+
+    pub const VT_TIMESTAMP: flatbuffers::VOffsetT = 4;
+
+  #[inline]
+  pub fn timestamp(&self) -> u64 {
+    self._tab.get::<u64>(Time::VT_TIMESTAMP, Some(0)).unwrap()
+  }
+}
+
+pub struct TimeArgs {
+    pub timestamp: u64,
+}
+impl<'a> Default for TimeArgs {
+    #[inline]
+    fn default() -> Self {
+        TimeArgs {
+            timestamp: 0,
+        }
+    }
+}
+pub struct TimeBuilder<'a: 'b, 'b> {
+  fbb_: &'b mut flatbuffers::FlatBufferBuilder<'a>,
+  start_: flatbuffers::WIPOffset<flatbuffers::TableUnfinishedWIPOffset>,
+}
+impl<'a: 'b, 'b> TimeBuilder<'a, 'b> {
+  #[inline]
+  pub fn add_timestamp(&mut self, timestamp: u64) {
+    self.fbb_.push_slot::<u64>(Time::VT_TIMESTAMP, timestamp, 0);
+  }
+  #[inline]
+  pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a>) -> TimeBuilder<'a, 'b> {
+    let start = _fbb.start_table();
+    TimeBuilder {
+      fbb_: _fbb,
+      start_: start,
+    }
+  }
+  #[inline]
+  pub fn finish(self) -> flatbuffers::WIPOffset<Time<'a>> {
+    let o = self.fbb_.end_table(self.start_);
+    flatbuffers::WIPOffset::new(o.value())
+  }
+}
+
 #[inline]
 pub fn get_root_as_sync_message<'a>(buf: &'a [u8]) -> SyncMessage<'a> {
   flatbuffers::get_root::<SyncMessage<'a>>(buf)

src/cli/run_impl.rs

@@ -15,7 +15,9 @@ use ckb_shared::cachedb::CacheDB;
 use ckb_shared::index::ChainIndex;
 use ckb_shared::shared::{ChainProvider, Shared, SharedBuilder};
 use ckb_shared::store::ChainKVStore;
-use ckb_sync::{Relayer, Synchronizer, RELAY_PROTOCOL_ID, SYNC_PROTOCOL_ID};
+use ckb_sync::{
+    NetTimeProtocol, Relayer, Synchronizer, RELAY_PROTOCOL_ID, SYNC_PROTOCOL_ID, TIME_PROTOCOL_ID,
+};
 use crypto::secp::Generator;
 use log::info;
 use numext_fixed_hash::H256;
@@ -64,6 +66,8 @@ pub fn run(setup: Setup) {
         synchronizer.peers(),
     ));
 
+    let net_time_checker = Arc::new(NetTimeProtocol::default());
+
     let network_config = NetworkConfig::from(setup.configs.network);
     let protocol_base_name = "ckb";
     let protocols = vec![
@@ -79,6 +83,12 @@ pub fn run(setup: Setup) {
             RELAY_PROTOCOL_ID,
             &[1][..],
         ),
+        CKBProtocol::new(
+            protocol_base_name.to_string(),
+            net_time_checker as Arc<_>,
+            TIME_PROTOCOL_ID,
+            &[1][..],
+        ),
     ];
     let network = Arc::new(
         NetworkService::run_in_thread(&network_config, protocols)

sync/src/lib.rs

@@ -4,6 +4,7 @@
 //! https://github.com/nervosnetwork/rfcs/tree/master/rfcs/0000-block-sync-protocol
 
 mod config;
+mod net_time_checker;
 mod relayer;
 mod synchronizer;
 mod types;
@@ -12,6 +13,7 @@ mod types;
 mod tests;
 
 pub use crate::config::Config;
+pub use crate::net_time_checker::NetTimeProtocol;
 pub use crate::relayer::Relayer;
 pub use crate::synchronizer::Synchronizer;
 
@@ -29,6 +31,7 @@ pub const BLOCK_DOWNLOAD_WINDOW: u64 = 1024;
 pub const PER_FETCH_BLOCK_LIMIT: usize = 128;
 pub const SYNC_PROTOCOL_ID: ProtocolId = *b"syn";
 pub const RELAY_PROTOCOL_ID: ProtocolId = *b"rel";
+pub const TIME_PROTOCOL_ID: ProtocolId = *b"tim";
 
 //  Timeout = base + per_header * (expected number of headers)
 pub const HEADERS_DOWNLOAD_TIMEOUT_BASE: u64 = 15 * 60 * 1000; // 15 minutes