validatord: moved tx protocol and json rpc to main p2p subnet, protocol_sync.rs: upon receiving finalized block, node removes blocks txs from unconfirmed_txs, /src/consensus/state.rs: decoupled removal of unconfirmed txs

script/research/nodes-tool/src/main.rs

@@ -333,7 +333,6 @@ struct StateInfo {
     _id: u64,
     _consensus: ConsensusInfo,
     _blockchain: BlockchainInfo,
-    _unconfirmed_txs: Vec<Tx>,
 }
 
 impl StateInfo {
@@ -341,8 +340,7 @@ impl StateInfo {
         let _id = state.id;
         let _consensus = ConsensusInfo::new(&state.consensus);
         let _blockchain = BlockchainInfo::new(&state.blockchain);
-        let _unconfirmed_txs = state.unconfirmed_txs.clone();
-        StateInfo { _id, _consensus, _blockchain, _unconfirmed_txs }
+        StateInfo { _id, _consensus, _blockchain }
     }
 }
 

script/research/validatord/src/main.rs

@@ -268,6 +268,8 @@ async fn proposal_task(p2p: net::P2pPtr, state: ValidatorStatePtr) {
             }
         };
 
+        info!("tt: {:?}", state.read().unwrap().unconfirmed_txs);
+
         // Node waits untile next epoch
         let seconds_until_next_epoch = state.read().unwrap().next_epoch_start();
         info!("Waiting for next epoch({:?} sec)...", seconds_until_next_epoch);
@@ -322,6 +324,15 @@ async fn start(executor: Arc<Executor<'_>>, opts: &Opt) -> Result<()> {
         })
         .await;
 
+    // Adding ProtocolTx to the registry
+    let state2 = state.clone();
+    registry
+        .register(net::SESSION_ALL, move |channel, main_p2p| {
+            let state = state2.clone();
+            async move { ProtocolTx::init(channel, state, main_p2p).await }
+        })
+        .await;
+
     // Performs seed session
     main_p2p.clone().start(executor.clone()).await?;
     // Actual main p2p session
@@ -335,6 +346,18 @@ async fn start(executor: Arc<Executor<'_>>, opts: &Opt) -> Result<()> {
         })
         .detach();
 
+    // RPC interface
+    let ex2 = executor.clone();
+    let ex3 = ex2.clone();
+    let rpc_interface = Arc::new(JsonRpcInterface {
+        state: state.clone(),
+        p2p: main_p2p.clone(),
+        _rpc_listen_addr: opts.rpc,
+    });
+    executor
+        .spawn(async move { listen_and_serve(rpc_server_config, rpc_interface, ex3).await })
+        .detach();
+
     // Node starts syncing
     let state2 = state.clone();
     syncing_task(main_p2p.clone(), state2).await?;
@@ -343,15 +366,6 @@ async fn start(executor: Arc<Executor<'_>>, opts: &Opt) -> Result<()> {
     let consensus_p2p = net::P2p::new(consensus_subnet_settings).await;
     let registry = consensus_p2p.protocol_registry();
 
-    // Adding ProtocolTx to the registry
-    let state2 = state.clone();
-    registry
-        .register(net::SESSION_ALL, move |channel, consensus_p2p| {
-            let state = state2.clone();
-            async move { ProtocolTx::init(channel, state, consensus_p2p).await }
-        })
-        .await;
-
     // Adding PropotolVote to the registry
     let p2p = main_p2p.clone();
     let state2 = state.clone();
@@ -403,18 +417,6 @@ async fn start(executor: Arc<Executor<'_>>, opts: &Opt) -> Result<()> {
         })
         .detach();
 
-    // RPC interface
-    let ex2 = executor.clone();
-    let ex3 = ex2.clone();
-    let rpc_interface = Arc::new(JsonRpcInterface {
-        state: state.clone(),
-        p2p: consensus_p2p.clone(),
-        _rpc_listen_addr: opts.rpc,
-    });
-    executor
-        .spawn(async move { listen_and_serve(rpc_server_config, rpc_interface, ex3).await })
-        .detach();
-
     proposal_task(consensus_p2p, state).await;
 
     Ok(())

script/research/validatord/src/protocols/protocol_sync.rs

@@ -86,11 +86,13 @@ impl ProtocolSync {
             // Commented for now, as to not mess consensus testing.
             // (Don't forget to remove _ from _p2p)
             /*
-            // Node stores finalized block, if it doesn't exists (checking by slot).
+            // Node stores finalized block, if it doesn't exists (checking by slot),
+            // and removes its transactions from the unconfirmed_txs vector.
             // Extra validations can be added here.
             let info_copy = (*info).clone();
             if !self.state.read().unwrap().blockchain.has_block(&info_copy)? {
                 self.state.write().unwrap().blockchain.add_by_info(info_copy.clone())?;
+                self.state.write().unwrap().remove_txs(info_copy.txs.clone())?;
                 self.p2p.broadcast(info_copy).await?;
             }
             */

script/research/validatord/src/protocols/protocol_tx.rs

@@ -49,6 +49,8 @@ impl ProtocolTx {
                 tx
             );
             let tx_copy = (*tx).clone();
+
+            // Nodes use unconfirmed_txs vector as seen_txs pool.
             if self.state.write().unwrap().append_tx(tx_copy.clone()) {
                 self.p2p.broadcast(tx_copy).await?;
             }

src/consensus/state.rs

@@ -440,6 +440,17 @@ impl ValidatorState {
         Ok(None)
     }
 
+    /// Note removes provided transactions vector, from unconfirmed_txs, if they exist.
+    pub fn remove_txs(&mut self, transactions: Vec<Tx>) -> Result<()> {
+        for tx in transactions {
+            if let Some(pos) = self.unconfirmed_txs.iter().position(|txs| *txs == tx) {
+                self.unconfirmed_txs.remove(pos);
+            }
+        }
+
+        Ok(())
+    }
+
     /// Provided an index, node checks if chain can be finalized.
     /// Consensus finalization logic: If node has observed the notarization of 3 consecutive
     /// proposals in a fork chain, it finalizes (appends to canonical blockchain) all proposals up to the middle block.
@@ -463,15 +474,11 @@ impl ValidatorState {
                 for proposal in &mut chain.proposals[..(consecutive - 1)] {
                     proposal.sm.finalized = true;
                     finalized.push(proposal.clone());
-                    for tx in proposal.txs.clone() {
-                        if let Some(pos) = self.unconfirmed_txs.iter().position(|txs| *txs == tx) {
-                            self.unconfirmed_txs.remove(pos);
-                        }
-                    }
                 }
                 chain.proposals.drain(0..(consecutive - 1));
                 for proposal in &finalized {
                     self.blockchain.add_by_proposal(proposal.clone())?;
+                    self.remove_txs(proposal.txs.clone())?;
                     to_broadcast.push(BlockInfo::new(
                         proposal.st,
                         proposal.sl,