Bridge/metrics polishing

bridges/centralized-ethereum/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2018"
 [dependencies]
 actix = { version = "0.13.0", default-features = false }
 async-jsonrpc-client = { git = "https://github.com/witnet/async-jsonrpc-client", features = ["tcp"], branch = "fix-tcp-leak" }
+chrono = "0.4.38"
 ctrlc = "3.1.3"
 env_logger = "0.9.0"
 envy = "0.4"

bridges/centralized-ethereum/src/actors/dr_database.rs

@@ -179,6 +179,13 @@ impl Message for SetDrState {
     type Result = Result<(), ()>;
 }
 
+/// Count number of data requests in given state
+pub struct CountDrsPerState;
+
+impl Message for CountDrsPerState {
+    type Result = Result<(u64, u64, u64, u64), ()>;
+}
+
 impl Handler<SetDrInfoBridge> for DrDatabase {
     type Result = ();
 
@@ -274,6 +281,26 @@ impl Handler<SetDrState> for DrDatabase {
     }
 }
 
+impl Handler<CountDrsPerState> for DrDatabase {
+    type Result = Result<(u64, u64, u64, u64), ()>;
+
+    fn handle(&mut self, _msg: CountDrsPerState, _ctx: &mut Self::Context) -> Self::Result {
+        Ok(self.dr.iter().fold(
+            (0u64, 0u64, 0u64, 0u64),
+            |(mut drs_new, mut drs_pending, mut drs_finished, mut drs_dismissed),
+             (_dr_id, dr_info)| {
+                match dr_info.dr_state {
+                    DrState::New => drs_new += 1,
+                    DrState::Pending => drs_pending += 1,
+                    DrState::Finished => drs_finished += 1,
+                    DrState::Dismissed => drs_dismissed += 1,
+                };
+                (drs_new, drs_pending, drs_finished, drs_dismissed)
+            },
+        ))
+    }
+}
+
 /// Required trait for being able to retrieve DrDatabase address from system registry
 impl actix::Supervised for DrDatabase {}
 

bridges/centralized-ethereum/src/actors/dr_sender/tests.rs

@@ -4,14 +4,14 @@ use witnet_data_structures::chain::{RADAggregate, RADRequest, RADRetrieve, RADTa
 #[test]
 fn deserialize_empty_dr() {
     // An empty data request is invalid with error 0xE0: BridgeMalformedRequest
-    let err = deserialize_and_validate_dr_bytes(&[], 1).unwrap_err();
+    let err = deserialize_and_validate_dr_bytes(&[], 1, 1).unwrap_err();
     assert_eq!(err.encode_cbor(), vec![216, 39, 129, 24, 224]);
 }
 
 #[test]
 fn deserialize_dr_not_protobuf() {
     // A malformed data request is invalid with error 0xE0: BridgeMalformedRequest
-    let err = deserialize_and_validate_dr_bytes(&[1, 2, 3, 4], 1).unwrap_err();
+    let err = deserialize_and_validate_dr_bytes(&[1, 2, 3, 4], 1, 1).unwrap_err();
     assert_eq!(err.encode_cbor(), vec![216, 39, 129, 24, 224]);
 }
 
@@ -55,7 +55,7 @@ fn deserialize_dr_high_value() {
     let dro_bytes = dro.to_pb_bytes().unwrap();
     // Setting the maximum allowed value to 1 nanowit below that will result in an error 0xE1:
     // BridgePoorIncentives
-    let err = deserialize_and_validate_dr_bytes(&dro_bytes, total_value - 1).unwrap_err();
+    let err = deserialize_and_validate_dr_bytes(&dro_bytes, total_value - 1, 1).unwrap_err();
     assert_eq!(err.encode_cbor(), vec![216, 39, 129, 24, 225]);
 }
 
@@ -78,7 +78,7 @@ fn deserialize_dr_collateral_one_nanowit() {
     assert_eq!(total_value, 20_000_000);
 
     let dro_bytes = dro.to_pb_bytes().unwrap();
-    let err = deserialize_and_validate_dr_bytes(&dro_bytes, total_value).unwrap_err();
+    let err = deserialize_and_validate_dr_bytes(&dro_bytes, total_value, 1).unwrap_err();
     assert_eq!(err.encode_cbor(), vec![216, 39, 129, 24, 224]);
 }
 
@@ -95,7 +95,7 @@ fn deserialize_dr_value_overflow() {
     };
 
     let dro_bytes = dro.to_pb_bytes().unwrap();
-    let err = deserialize_and_validate_dr_bytes(&dro_bytes, 1).unwrap_err();
+    let err = deserialize_and_validate_dr_bytes(&dro_bytes, 1, 1).unwrap_err();
     assert_eq!(err.encode_cbor(), vec![216, 39, 129, 24, 224]);
 }
 
@@ -115,6 +115,6 @@ fn deserialize_and_validate_dr_bytes_wip_0022() {
     let dro_bytes = dro.to_pb_bytes().unwrap();
     let witnet_dr_max_value_nanowits = 100_000_000_000;
     let err =
-        deserialize_and_validate_dr_bytes(&dro_bytes, witnet_dr_max_value_nanowits).unwrap_err();
+        deserialize_and_validate_dr_bytes(&dro_bytes, witnet_dr_max_value_nanowits, 1).unwrap_err();
     assert_eq!(err.encode_cbor(), vec![216, 39, 129, 24, 224]);
 }

bridges/centralized-ethereum/src/actors/eth_poller.rs

@@ -113,7 +113,7 @@ impl EthPoller {
                     );
                     last_dr_id = skip_first;
                 }
-                while last_dr_id < next_dr_id {
+                while last_dr_id + 1 < next_dr_id {
                     let init_index = usize::try_from(last_dr_id + 1).unwrap();
                     let last_index = match next_dr_id.cmp(&(last_dr_id + max_batch_size)) {
                         std::cmp::Ordering::Greater => {

bridges/centralized-ethereum/src/actors/mod.rs

@@ -12,3 +12,6 @@ pub mod eth_poller;
 
 /// wit_poller actor module
 pub mod wit_poller;
+
+/// watch_dog actor module
+pub mod watch_dog;