Merge pull request #6412 from jferrant/hotfix/add-stackerdb-request-timeout

Hotfix/add stackerdb request timeout

Author: wileyj

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to the versioning scheme outlined in the [README.md](README.md).
 
+## [Unreleased]
+
+### Added
+
+- Add `stackerdb_timeout_secs` to miner config for limiting duration of StackerDB HTTP requests.
+
 ## [3.2.0.0.1]
 ### Added
 

libsigner/src/session.rs

@@ -16,6 +16,7 @@
 
 use std::net::TcpStream;
 use std::str;
+use std::time::Duration;
 
 use clarity::vm::types::QualifiedContractIdentifier;
 use libstackerdb::{
@@ -103,22 +104,34 @@ pub struct StackerDBSession {
     pub stackerdb_contract_id: QualifiedContractIdentifier,
     /// connection to the replica
     sock: Option<TcpStream>,
+    /// The timeout applied to HTTP read and write operations
+    socket_timeout: Duration,
 }
 
 impl StackerDBSession {
     /// instantiate but don't connect
-    pub fn new(host: &str, stackerdb_contract_id: QualifiedContractIdentifier) -> StackerDBSession {
+    pub fn new(
+        host: &str,
+        stackerdb_contract_id: QualifiedContractIdentifier,
+        socket_timeout: Duration,
+    ) -> StackerDBSession {
         StackerDBSession {
             host: host.to_owned(),
             stackerdb_contract_id,
             sock: None,
+            socket_timeout,
         }
     }
 
     /// connect or reconnect to the node
     fn connect_or_reconnect(&mut self) -> Result<(), RPCError> {
         debug!("connect to {}", &self.host);
-        self.sock = Some(TcpStream::connect(&self.host)?);
+        let sock = TcpStream::connect(&self.host)?;
+        // Make sure we don't hang forever if for some reason our node does not
+        // respond as expected such as failing to properly close the connection
+        sock.set_read_timeout(Some(self.socket_timeout))?;
+        sock.set_write_timeout(Some(self.socket_timeout))?;
+        self.sock = Some(sock);
         Ok(())
     }
 
@@ -251,11 +264,49 @@ impl SignerSession for StackerDBSession {
     /// upload a chunk
     fn put_chunk(&mut self, chunk: &StackerDBChunkData) -> Result<StackerDBChunkAckData, RPCError> {
         let body =
-            serde_json::to_vec(chunk).map_err(|e| RPCError::Deserialize(format!("{:?}", &e)))?;
+            serde_json::to_vec(chunk).map_err(|e| RPCError::Deserialize(format!("{e:?}")))?;
         let path = stackerdb_post_chunk_path(self.stackerdb_contract_id.clone());
         let resp_bytes = self.rpc_request("POST", &path, Some("application/json"), &body)?;
         let ack: StackerDBChunkAckData = serde_json::from_slice(&resp_bytes)
-            .map_err(|e| RPCError::Deserialize(format!("{:?}", &e)))?;
+            .map_err(|e| RPCError::Deserialize(format!("{e:?}")))?;
         Ok(ack)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::io::Write;
+    use std::net::TcpListener;
+    use std::thread;
+
+    use super::*;
+
+    #[test]
+    fn socket_timeout_works_as_expected() {
+        let listener = TcpListener::bind("127.0.0.1:0").expect("bind failed");
+        let addr = listener.local_addr().unwrap();
+
+        let short_timeout = Duration::from_millis(200);
+        thread::spawn(move || {
+            if let Ok((mut stream, _)) = listener.accept() {
+                // Sleep long enough so the client should hit its timeout
+                std::thread::sleep(short_timeout * 2);
+                let _ = stream.write_all(b"HTTP/1.1 200 OK\r\n\r\n");
+            }
+        });
+
+        let contract_id = QualifiedContractIdentifier::transient();
+        let mut session = StackerDBSession::new(&addr.to_string(), contract_id, short_timeout);
+
+        session.connect_or_reconnect().expect("connect failed");
+
+        // This should fail due to the timeout
+        let result = session.rpc_request("GET", "/", None, &[]);
+        match result {
+            Err(RPCError::IO(e)) => {
+                assert_eq!(e.kind(), std::io::ErrorKind::WouldBlock);
+            }
+            other => panic!("expected timeout error, got {other:?}"),
+        }
+    }
+}

stacks-node/src/nakamoto_node/miner.rs

@@ -1106,7 +1106,11 @@ impl BlockMinerThread {
             NakamotoNodeError::MinerConfigurationFailed("Failed to get RPC loopback socket")
         })?;
         let miners_contract_id = boot_code_id(MINERS_NAME, chain_state.mainnet);
-        let mut miners_session = StackerDBSession::new(&rpc_socket.to_string(), miners_contract_id);
+        let mut miners_session = StackerDBSession::new(
+            &rpc_socket.to_string(),
+            miners_contract_id,
+            self.config.miner.stackerdb_timeout,
+        );
 
         if Self::fault_injection_skip_block_push() {
             warn!(

stacks-node/src/nakamoto_node/signer_coordinator.rs

@@ -107,7 +107,11 @@ impl SignerCoordinator {
             .get_rpc_loopback()
             .ok_or_else(|| ChainstateError::MinerAborted)?;
         let miners_contract_id = boot_code_id(MINERS_NAME, is_mainnet);
-        let miners_session = StackerDBSession::new(&rpc_socket.to_string(), miners_contract_id);
+        let miners_session = StackerDBSession::new(
+            &rpc_socket.to_string(),
+            miners_contract_id,
+            config.miner.stackerdb_timeout,
+        );
 
         // build a BTreeMap of the various timeout steps
         let mut block_rejection_timeout_steps = BTreeMap::<u32, Duration>::new();

stacks-node/src/nakamoto_node/stackerdb_listener.rs

@@ -179,8 +179,11 @@ impl StackerDBListener {
             .node
             .get_rpc_loopback()
             .ok_or_else(|| ChainstateError::MinerAborted)?;
-        let mut signers_session =
-            StackerDBSession::new(&rpc_socket.to_string(), signers_contract_id.clone());
+        let mut signers_session = StackerDBSession::new(
+            &rpc_socket.to_string(),
+            signers_contract_id.clone(),
+            config.miner.stackerdb_timeout,
+        );
         let entries: Vec<_> = signer_entries.values().cloned().collect();
         let parsed_entries = SignerEntries::parse(config.is_mainnet(), &entries)
             .expect("FATAL: could not parse retrieved signer entries");

stacks-node/src/neon_node.rs

@@ -2302,8 +2302,11 @@ impl BlockMinerThread {
     /// Only used in mock signing to determine if the peer info view was already signed across
     fn mock_block_exists(&self, peer_info: &PeerInfo) -> bool {
         let miner_contract_id = boot_code_id(MINERS_NAME, self.config.is_mainnet());
-        let mut miners_stackerdb =
-            StackerDBSession::new(&self.config.node.rpc_bind, miner_contract_id);
+        let mut miners_stackerdb = StackerDBSession::new(
+            &self.config.node.rpc_bind,
+            miner_contract_id,
+            self.config.miner.stackerdb_timeout,
+        );
         let miner_slot_ids: Vec<_> = (0..MINER_SLOT_COUNT * 2).collect();
         if let Ok(messages) = miners_stackerdb.get_latest_chunks(&miner_slot_ids) {
             for message in messages.into_iter().flatten() {
@@ -2379,8 +2382,11 @@ impl BlockMinerThread {
         let stackerdbs = StackerDBs::connect(&self.config.get_stacker_db_file_path(), false)
             .map_err(|e| e.to_string())?;
         let miner_contract_id = boot_code_id(MINERS_NAME, self.config.is_mainnet());
-        let mut miners_stackerdb =
-            StackerDBSession::new(&self.config.node.rpc_bind, miner_contract_id);
+        let mut miners_stackerdb = StackerDBSession::new(
+            &self.config.node.rpc_bind,
+            miner_contract_id,
+            self.config.miner.stackerdb_timeout,
+        );
         let miner_db = MinerDB::open_with_config(&self.config).map_err(|e| e.to_string())?;
 
         SignerCoordinator::send_miners_message(

stacks-node/src/tests/nakamoto_integrations.rs

@@ -486,7 +486,11 @@ pub fn get_latest_block_proposal(
     let miner_ranges = stackerdb_conf.signer_ranges();
     let latest_miner = usize::from(miner_info.get_latest_winner_index());
     let miner_contract_id = boot_code_id(MINERS_NAME, false);
-    let mut miners_stackerdb = StackerDBSession::new(&conf.node.rpc_bind, miner_contract_id);
+    let mut miners_stackerdb = StackerDBSession::new(
+        &conf.node.rpc_bind,
+        miner_contract_id,
+        Duration::from_secs(30),
+    );
 
     let mut proposed_blocks: Vec<_> = stackerdb_conf
         .signers
@@ -12660,29 +12664,28 @@ fn write_signer_update(
 ) {
     let signers_contract_id =
         MessageSlotID::StateMachineUpdate.stacker_db_contract(false, reward_cycle);
-    let mut session = StackerDBSession::new(&conf.node.rpc_bind, signers_contract_id);
+    let mut session = StackerDBSession::new(
+        &conf.node.rpc_bind,
+        signers_contract_id,
+        Duration::from_secs(30),
+    );
     let message = SignerMessageV0::StateMachineUpdate(update);
 
-    // Submit the block proposal to the signers slot
-    let mut accepted = false;
+    // Submit the update to the signers slot
     let mut version = 0;
-    let start = Instant::now();
-    while !accepted {
+    wait_for(timeout.as_secs(), || {
         let mut chunk =
             StackerDBChunkData::new(signer_slot_id, version, message.serialize_to_vec());
         chunk
             .sign(&signer_sk)
             .expect("Failed to sign message chunk");
         debug!("Produced a signature: {:?}", chunk.sig);
         let result = session.put_chunk(&chunk).expect("Failed to put chunk");
-        accepted = result.accepted;
         version += 1;
         debug!("Test Put Chunk ACK: {result:?}");
-        assert!(
-            start.elapsed() < timeout,
-            "Timed out waiting for signer state update to be accepted"
-        );
-    }
+        Ok(result.accepted)
+    })
+    .expect("Failed to accept signer state update");
 }
 
 /// Test SIP-031 activation

stacks-node/src/tests/signer/mod.rs

@@ -1462,6 +1462,7 @@ impl<Z: SpawnedSignerTrait> SignerTest<Z> {
             self.get_current_reward_cycle(),
             SignerSlotID(0), // We are just reading so again, don't care about index.
             SignerDb::new(":memory:").unwrap(),
+            Duration::from_secs(30),
         );
         let mut latest_msgs = StackerDB::get_messages(
             stackerdb
@@ -1526,6 +1527,7 @@ impl<Z: SpawnedSignerTrait> SignerTest<Z> {
             reward_cycle,
             SignerSlotID(0), // We are just reading so again, don't care about index.
             SignerDb::new(":memory:").unwrap(), // also don't care about the signer db for version tracking
+            Duration::from_secs(30),
         )
     }
 
@@ -1570,6 +1572,7 @@ impl<Z: SpawnedSignerTrait> SignerTest<Z> {
                 .expect("Failed to get signer slot id")
                 .expect("Signer does not have a slot id"),
             SignerDb::new(":memory:").unwrap(),
+            Duration::from_secs(30),
         );
 
         let signature = private_key

stacks-node/src/tests/signer/v0.rs

@@ -473,8 +473,11 @@ impl SignerTest<SpawnedSigner> {
     /// Propose a block to the signers
     fn propose_block(&self, block: NakamotoBlock, timeout: Duration) {
         let miners_contract_id = boot_code_id(MINERS_NAME, false);
-        let mut session =
-            StackerDBSession::new(&self.running_nodes.conf.node.rpc_bind, miners_contract_id);
+        let mut session = StackerDBSession::new(
+            &self.running_nodes.conf.node.rpc_bind,
+            miners_contract_id,
+            self.running_nodes.conf.miner.stackerdb_timeout,
+        );
         let burn_height = self
             .running_nodes
             .btc_regtest_controller
@@ -17774,8 +17777,11 @@ fn miner_stackerdb_version_rollover() {
         max_chunk.slot_version
     );
 
-    let mut stackerdb =
-        StackerDBSession::new(&conf_2.node.rpc_bind, boot_code_id(MINERS_NAME, false));
+    let mut stackerdb = StackerDBSession::new(
+        &conf_2.node.rpc_bind,
+        boot_code_id(MINERS_NAME, false),
+        conf_2.miner.stackerdb_timeout,
+    );
 
     let proposals_before = miners.get_primary_proposals_submitted().get();
 

stacks-signer/CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to the versioning scheme outlined in the [README.md](README.md).
 
+## [3.2.0.0.1.1]
+
+### Added
+
+- Introduced `stackerdb_timeout_secs`: config option to set the maximum time (in seconds) the signer will wait for StackerDB HTTP requests to complete.
+
 ## [3.2.0.0.1.0]
 
 ### Changed