feat(spv): flush header index on shutdown and after header sync (#197)

* feat(spv): flush headers on shutdown

* move fn lower in the impl

* refactor: `MultiPeerNetworkManager` -> `PeerNetworkManager` (#184)

* refactor: `MultiPeerNetworkManager` -> `PeerNetworkManager`

* Fix formatting and apply review

* feat: Update ffi headers (#183)

* feat(spv): broadcast transaction (#180)

* fix: Fix `PeerNetworkManager` cast in `broadcast_transaction` (#185)

PR #180 used `MultiPeerNetworkManager`, this was renamed in #183 which was merged before #180 but after its CI run.

* fix: Use non-blocking `TcpStream` in `dash-spv::network::TcpConnection` (#188)

* refactor: Improve SPV shutdown handling with `CancellationToken` (#187)

* refactor: `TcpConnection` -> `Peer` and `ConnectionPool` -> `PeerPool` (#190)

* fix: Locking issue after #190 (#191)

#190 removed the read timeouts of the `Peer::receive_message` which currently leads to a lockup of the peer because the write lock is held while waiting for the message. Needs some more refactoring but this works for now.

* fix: More follow-up to #190 (#193)

The sleep timeout branch introduced in #191 returns an `Err(NetworkError::Timeout)` which leads to a misbehavior update below in the `msg_result` match and eventually in a peer ban. This shouldn't happen because the `sleep` timing out only means that there is no data available right now. Instead, it now returns `Ok(None)` which will just keep things going.

* flush after mn list sync too

* move flush to after header sync instead of mnlist

* fix

---------

Co-authored-by: Kevin Rombach <35775977+xdustinface@users.noreply.github.com>

Author: pauldelucia

dash-spv/src/storage/disk/state.rs

@@ -436,6 +436,13 @@ impl DiskStorageManager {
 
         // Shutdown background worker
         if let Some(tx) = self.worker_tx.take() {
+            // Save the header index before shutdown
+            let index = self.header_hash_index.read().await.clone();
+            let _ = tx
+                .send(super::manager::WorkerCommand::SaveIndex {
+                    index,
+                })
+                .await;
             let _ = tx.send(super::manager::WorkerCommand::Shutdown).await;
         }
 
@@ -701,6 +708,29 @@ mod tests {
     use dashcore::{block::Version, pow::CompactTarget};
     use tempfile::TempDir;
 
+    fn build_headers(count: usize) -> Vec<BlockHeader> {
+        let mut headers = Vec::with_capacity(count);
+        let mut prev_hash = BlockHash::from_byte_array([0u8; 32]);
+
+        for i in 0..count {
+            let header = BlockHeader {
+                version: Version::from_consensus(1),
+                prev_blockhash: prev_hash,
+                merkle_root: dashcore::hashes::sha256d::Hash::from_byte_array(
+                    [(i % 255) as u8; 32],
+                )
+                .into(),
+                time: 1 + i as u32,
+                bits: CompactTarget::from_consensus(0x1d00ffff),
+                nonce: i as u32,
+            };
+            prev_hash = header.block_hash();
+            headers.push(header);
+        }
+
+        headers
+    }
+
     #[tokio::test]
     async fn test_sentinel_headers_not_returned() -> Result<(), Box<dyn std::error::Error>> {
         // Create a temporary directory for the test
@@ -859,4 +889,28 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn test_shutdown_flushes_index() -> Result<(), Box<dyn std::error::Error>> {
+        let temp_dir = TempDir::new()?;
+        let base_path = temp_dir.path().to_path_buf();
+        let headers = build_headers(11_000);
+        let last_hash = headers.last().unwrap().block_hash();
+
+        {
+            let mut storage = DiskStorageManager::new(base_path.clone()).await?;
+
+            storage.store_headers(&headers[..10_000]).await?;
+            super::super::segments::save_dirty_segments(&storage).await?;
+
+            storage.store_headers(&headers[10_000..]).await?;
+            storage.shutdown().await?;
+        }
+
+        let storage = DiskStorageManager::new(base_path).await?;
+        let height = storage.get_header_height_by_hash(&last_hash).await?;
+        assert_eq!(height, Some(10_999));
+
+        Ok(())
+    }
 }

dash-spv/src/sync/sequential/message_handlers.rs

@@ -392,8 +392,6 @@ impl<
 
         if should_transition {
             self.transition_to_next_phase(storage, network, "Headers sync complete").await?;
-
-            // Execute the next phase
             self.execute_current_phase(network, storage).await?;
         }