use a patched codec

contains more verbose information about decoding errors

Author: insipx

Cargo.lock

@@ -70,6 +70,7 @@ jsonrpc-client-transports = { git = "https://github.com/paritytech/jsonrpc", bra
 jsonrpc-core = { git = "https://github.com/paritytech/jsonrpc", branch = "master", features = ["arbitrary_precision"] }
 jsonrpc-derive = { git = "https://github.com/paritytech/jsonrpc", branch = "master" }
 jsonrpc-pubsub = { git = "https://github.com/paritytech/jsonrpc", branch = "master" }
+codec = { git = "https://github.com/insipx/parity-scale-codec", branch = "insipx/debugging", package = "parity-scale-codec" }
 
 [dev-dependencies]
 node-runtime = { git = "https://github.com/paritytech/substrate", package="node-runtime", branch="polkadot-master" }

Cargo.toml

@@ -143,7 +143,6 @@ impl ExtractCall for CallWrapper {
             Call::Registrar(call) => {
                 (Module::Custom("Registrar".into()), Box::new(RegistrarCallWrapper(call.clone())))
             },
-            //ElectionsPhragmen
             c @ _ => {
                 warn!("Call Not Handled: {:?}", c);
                 (Module::NotHandled, Box::new(NotHandled))

polkadot-archive/src/main.rs

@@ -17,46 +17,44 @@
 //! Spawning of all tasks happens in this module
 //! Nowhere else is anything ever spawned
 
-use log::*;
-use tokio::runtime::Runtime;
-use substrate_rpc_primitives::number::NumberOrHex;
 use futures::{
-    Future, Stream,
+    future::{self, loop_fn, Loop},
     sync::mpsc::{self, UnboundedReceiver, UnboundedSender},
-    future::{self, loop_fn, Loop}
-};
-use substrate_primitives::{
-    U256,
-    storage::StorageKey,
-    twox_128
+    Future, Stream,
 };
+use log::*;
+use substrate_primitives::{storage::StorageKey, twox_128, U256};
+use substrate_rpc_primitives::number::NumberOrHex;
+use tokio::runtime::Runtime;
 
 use std::sync::Arc;
 
 use crate::{
     database::Database,
-    rpc::Rpc,
     error::Error as ArchiveError,
-    types::{System, Data}
+    rpc::Rpc,
+    types::{Data, System},
 };
 
 // with the hopeful and long-anticipated release of async-await
 pub struct Archive<T: System> {
     rpc: Arc<Rpc<T>>,
     db: Arc<Database>,
-    runtime: Runtime
+    runtime: Runtime,
 }
 
-impl<T> Archive<T> where T: System {
-
+impl<T> Archive<T>
+where
+    T: System,
+{
     pub fn new() -> Result<Self, ArchiveError> {
         let mut runtime = Runtime::new()?;
         let rpc = runtime.block_on(Rpc::<T>::new(url::Url::parse("ws://127.0.0.1:9944")?))?;
         let db = Database::new()?;
         let (rpc, db) = (Arc::new(rpc), Arc::new(db));
         debug!("METADATA: {}", rpc.metadata());
         debug!("KEYS: {:?}", rpc.keys());
-        Ok( Self { rpc, db, runtime })
+        Ok(Self { rpc, db, runtime })
     }
 
     pub fn run(mut self) -> Result<(), ArchiveError> {
@@ -65,45 +63,44 @@ impl<T> Archive<T> where T: System {
         self.runtime.spawn(
             self.rpc
                 .clone()
-                .subscribe_blocks(sender.clone()).map_err(|e| println!("{:?}", e))
-        );
-        self.runtime.spawn(
-            Self::sync(self.db.clone(), self.rpc.clone(), sender.clone())
-                .map(|_| ())
+                .subscribe_blocks(sender.clone())
+                .map_err(|e| println!("{:?}", e)),
         );
+        self.runtime
+            .spawn(Self::sync(self.db.clone(), self.rpc.clone(), sender.clone()).map(|_| ()));
         tokio::run(Self::handle_data(receiver, self.db.clone()));
         Ok(())
     }
 
     // TODO return a float between 0 and 1 corresponding to percent of database that is up-to-date?
     /// Verification task that ensures all blocks are in the database
-    fn sync(db: Arc<Database>,
-              rpc: Arc<Rpc<T>>,
-              sender: UnboundedSender<Data<T>>
-    ) -> impl Future<Item = Sync, Error = ()> + 'static
-    {
+    fn sync(
+        db: Arc<Database>,
+        rpc: Arc<Rpc<T>>,
+        sender: UnboundedSender<Data<T>>,
+    ) -> impl Future<Item = Sync, Error = ()> + 'static {
         loop_fn(Sync::new(), move |v| {
             let sender0 = sender.clone();
             v.sync(db.clone(), rpc.clone(), sender0.clone())
-             .and_then(move |(sync, done)| {
-                 if done {
-                     Ok(Loop::Break(sync))
-                 } else {
-                     Ok(Loop::Continue(sync))
-                 }
-             })
+                .and_then(move |(sync, done)| {
+                    if done {
+                        Ok(Loop::Break(sync))
+                    } else {
+                        Ok(Loop::Continue(sync))
+                    }
+                })
         })
     }
 
-    fn handle_data(receiver: UnboundedReceiver<Data<T>>,
-                    db: Arc<Database>,
-    ) -> impl Future<Item = (), Error = ()> + 'static
-    {
+    fn handle_data(
+        receiver: UnboundedReceiver<Data<T>>,
+        db: Arc<Database>,
+    ) -> impl Future<Item = (), Error = ()> + 'static {
         receiver.for_each(move |data| {
             match data {
                 Data::SyncProgress(missing_blocks) => {
                     println!("{} blocks missing", missing_blocks);
-                },
+                }
                 c => {
                     tokio::spawn(db.insert(c).map_err(|e| error!("{:?}", e)));
                 }
@@ -120,60 +117,64 @@ struct Sync {
 
 impl Sync {
     fn new() -> Self {
-        Self {
-            looped: 0,
-        }
+        Self { looped: 0 }
     }
 
-    fn sync<T>(self,
-                 db: Arc<Database>,
-                 rpc: Arc<Rpc<T>>,
-                 sender: UnboundedSender<Data<T>>,
+    fn sync<T>(
+        self,
+        db: Arc<Database>,
+        rpc: Arc<Rpc<T>>,
+        sender: UnboundedSender<Data<T>>,
     ) -> impl Future<Item = (Self, bool), Error = ()> + 'static
-        where T: System + std::fmt::Debug + 'static
+    where
+        T: System + std::fmt::Debug + 'static,
     {
         let (sender0, sender1) = (sender.clone(), sender.clone());
         let (rpc0, rpc1) = (rpc.clone(), rpc.clone());
         let looped = self.looped;
         info!("Looped: {}", looped);
 
-        let missing_blocks =
-            db.query_missing_blocks()
-              .and_then(move |blocks| {
-                  match sender0
-                      .unbounded_send(Data::SyncProgress(blocks.len()))
-                      .map_err(Into::into) {
-                          Ok(()) => (),
-                          Err(e) => return future::err(e)
-                      }
-                  future::ok(
-                      blocks
-                          .into_iter()
-                          .take(100_000) // just do 100K blocks at a time
-                          .map(|b| NumberOrHex::Hex(U256::from(b)))
-                          .collect::<Vec<NumberOrHex<T::BlockNumber>>>()
-                  )
-              }).and_then(move |blocks| {
-                  rpc0.batch_block_from_number(blocks, sender)
-                     .and_then(move |_| {
-                         let looped = looped + 1;
-                         future::ok((Self {looped}, false ))
-                     })
-              });
-
-        let missing_timestamps =
-            db.query_missing_timestamps::<T>()
-            .and_then(move |hashes| {
-                info!("Launching timestamp insertion thread for {} items", hashes.len());
-                let timestamp_key = b"Timestamp Now";
-                let storage_key = twox_128(timestamp_key);
-                let keys = std::iter::repeat(StorageKey(storage_key.to_vec()))
-                    .take(hashes.len())
-                    .collect::<Vec<StorageKey>>();
-                rpc1.batch_storage(sender1, keys, hashes)
+        let missing_blocks = db
+            .query_missing_blocks()
+            .and_then(move |blocks| {
+                match sender0
+                    .unbounded_send(Data::SyncProgress(blocks.len()))
+                    .map_err(Into::into)
+                {
+                    Ok(()) => (),
+                    Err(e) => return future::err(e),
+                }
+                future::ok(
+                    blocks
+                        .into_iter()
+                        .take(100_000) // just do 100K blocks at a time
+                        .map(|b| NumberOrHex::Hex(U256::from(b)))
+                        .collect::<Vec<NumberOrHex<T::BlockNumber>>>(),
+                )
+            })
+            .and_then(move |blocks| {
+                rpc0.batch_block_from_number(blocks, sender)
+                    .and_then(move |_| {
+                        let looped = looped + 1;
+                        future::ok((Self { looped }, false))
+                    })
             });
-        missing_timestamps.join(missing_blocks)
-                          .map_err(|e| error!("{:?}", e))
-                          .map(|(_, b)| b)
+
+        let missing_timestamps = db.query_missing_timestamps::<T>().and_then(move |hashes| {
+            info!(
+                "Launching timestamp insertion thread for {} items",
+                hashes.len()
+            );
+            let timestamp_key = b"Timestamp Now";
+            let storage_key = twox_128(timestamp_key);
+            let keys = std::iter::repeat(StorageKey(storage_key.to_vec()))
+                .take(hashes.len())
+                .collect::<Vec<StorageKey>>();
+            rpc1.batch_storage(sender1, keys, hashes)
+        });
+        missing_timestamps
+            .join(missing_blocks)
+            .map_err(|e| error!("{:?}", e))
+            .map(|(_, b)| b)
     }
 }

src/archive.rs

@@ -78,12 +78,13 @@ where
 
         if bytes[0] == 0x16 {
             let mut bytes = &bytes[1..];
+            log::trace!("BYTES WITHOUT 0x16: {:?}", bytes);
             let reg: Result<RegistrarCall<RuntimeT>, _> = Decode::decode(&mut bytes).map_err(|_| warn!("Did not decode"));
             log::trace!("REGISTRAR CALL: {:?}", reg);
         }
 
+        log::trace!("bytes read from decoding extrinsic: {:X?}", bytes);
         let function = Decode::decode(&mut bytes.as_slice()).map_err(|e| { warn!("Error decoding call"); e })?;
-
         Ok(Self { signature, function, version })
     }
 }

src/extrinsics.rs

@@ -75,14 +75,16 @@ pub trait System: Send + Sync + 'static + Debug {
 
     /// the Opaque Extrinsic Type
     type Extrinsic: Send
-        + Sync + ToDatabaseExtrinsic
+        + Sync
+        + ToDatabaseExtrinsic
         + Debug
         + Serialize
         + DeserializeOwned
         + Clone
         + Eq
         + PartialEq
         + Unpin;
+    // require Iter
 
     // type Block: BlockTrait + Encode + Decode + Debug;
     type Signature: Encode + Decode + Debug;