fix(headers2): Fix compressed headers protocol compatibility with Dash Core

PastaPastaPasta · claude · PastaPastaPasta · commit a81cf271a039 · 2025-12-09T16:18:45.000-06:00
This commit fixes critical incompatibilities between the Rust headers2 implementation and the C++ Dash Core reference implementation (DIP-0025). ## Issues Fixed ### Version Offset Encoding Mismatch - C++ uses offset=0 for "version not in cache" (full version present) - C++ uses offset=1-7 for "version at position offset-1 in cache" - Rust incorrectly used offset=7 for uncompressed, offset=0-6 for cached - Now matches C++ semantics exactly ### Version Cache Data Structure - C++ uses std::list with MRU (Most Recently Used) reordering - Rust used a circular buffer without MRU reordering - Changed to Vec<i32> with proper MRU behavior matching C++ ### Serialization/Deserialization - Fixed Decodable impl to read version when offset=0 (not offset=7) - Added MissingVersion error variant for proper error handling ## Changes ### dashcore (dash/src/network/message_headers2.rs) - Rewrote CompressionState to use Vec with MRU reordering - Fixed compress() to use offset=0 for uncompressed versions - Fixed decompress() to handle C++ offset semantics - Updated Decodable to read version when offset=0 - Added comprehensive tests for C++ compatibility ### dash-spv - Enabled headers2 in handshake negotiation - Enabled headers2 in sync manager - Fixed phase transition when receiving empty headers2 response - Re-enabled has_headers2_peer() check ## Testing - Added headers2_compatibility_test.rs with 12 tests verifying: - Version offset C++ semantics - MRU cache reordering behavior - Flag bit semantics - Serialization format compatibility - Cross-implementation compatibility - All existing tests pass 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
diff --git a/dash-spv/src/network/handshake.rs b/dash-spv/src/network/handshake.rs
@@ -252,8 +252,8 @@ impl HandshakeManager {
             .unwrap_or(Duration::from_secs(0))
             .as_secs() as i64;
 
-        // SPV client doesn't advertise any special services since headers2 is disabled
-        let services = ServiceFlags::NONE;
+        // Advertise headers2 support (NODE_HEADERS_COMPRESSED)
+        let services = ServiceFlags::NONE | NODE_HEADERS_COMPRESSED;
 
         // Parse the local address safely
         let local_addr = "127.0.0.1:0"
@@ -310,10 +310,13 @@ impl HandshakeManager {
 
     /// Negotiate headers2 support with the peer after handshake completion.
     async fn negotiate_headers2(&self, connection: &mut Peer) -> NetworkResult<()> {
-        // Headers2 is currently disabled due to protocol compatibility issues
-        // Always send SendHeaders regardless of peer support
-        tracing::info!("Headers2 is disabled - sending SendHeaders only");
-        connection.send_message(NetworkMessage::SendHeaders).await?;
+        if self.peer_supports_headers2() {
+            tracing::info!("Peer supports headers2 - sending SendHeaders2");
+            connection.send_message(NetworkMessage::SendHeaders2).await?;
+        } else {
+            tracing::info!("Peer does not support headers2 - sending SendHeaders");
+            connection.send_message(NetworkMessage::SendHeaders).await?;
+        }
         Ok(())
     }
 }
diff --git a/dash-spv/src/network/manager.rs b/dash-spv/src/network/manager.rs
@@ -423,19 +423,20 @@ impl PeerNetworkManager {
                                     headers.len()
                                 );
                                 // Check if peer supports headers2
-                                // TODO: Re-enable this warning once headers2 is fixed
-                                // Currently suppressed since headers2 is disabled
-                                /*
                                 let peer_guard = peer.read().await;
-                                if peer_guard.peer_info().services.map(|s| {
-                                    dashcore::network::constants::ServiceFlags::from(s).has(
-                                        dashcore::network::constants::ServiceFlags::from(2048u64)
-                                    )
-                                }).unwrap_or(false) {
+                                if peer_guard
+                                    .peer_info()
+                                    .services
+                                    .map(|s| {
+                                        dashcore::network::constants::ServiceFlags::from(s).has(
+                                            dashcore::network::constants::NODE_HEADERS_COMPRESSED,
+                                        )
+                                    })
+                                    .unwrap_or(false)
+                                {
                                     log::warn!("⚠️  Peer {} supports headers2 but sent regular headers - possible protocol issue", addr);
                                 }
                                 drop(peer_guard);
-                                */
                                 // Forward to client
                             }
                             NetworkMessage::Headers2(headers2) => {
@@ -1296,9 +1297,7 @@ impl NetworkManager for PeerNetworkManager {
     }
 
     async fn has_headers2_peer(&self) -> bool {
-        // Headers2 is currently disabled due to protocol compatibility issues
-        // TODO: Fix headers2 decompression before re-enabling
-        false
+        self.has_peer_with_service(dashcore::network::constants::NODE_HEADERS_COMPRESSED).await
     }
 
     async fn get_last_message_peer_id(&self) -> crate::types::PeerId {
diff --git a/dash-spv/src/sync/headers/manager.rs b/dash-spv/src/sync/headers/manager.rs
@@ -412,16 +412,8 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
                         hash
                     );
                     vec![hash]
-                } else if network.has_headers2_peer().await && !self.headers2_failed {
-                    // Check if this is genesis and we're using headers2
-                    let genesis_hash = self.config.network.known_genesis_block_hash();
-                    if genesis_hash == Some(hash) {
-                        tracing::info!("📍 Using empty locator for headers2 genesis sync");
-                        vec![]
-                    } else {
-                        vec![hash]
-                    }
                 } else {
+                    // Always include the hash in the locator - peer needs to know our position
                     vec![hash]
                 }
             }
@@ -462,9 +454,8 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
             getheaders_msg.stop_hash
         );
 
-        // Headers2 is currently disabled due to protocol compatibility issues
-        // TODO: Fix headers2 decompression before re-enabling
-        let use_headers2 = false; // Disabled until headers2 implementation is fixed
+        // Use headers2 if peer supports it and we haven't had failures
+        let use_headers2 = network.has_headers2_peer().await && !self.headers2_failed;
 
         // Log details about the request
         tracing::info!(
@@ -535,64 +526,50 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
         &mut self,
         headers2: dashcore::network::message_headers2::Headers2Message,
         peer_id: crate::types::PeerId,
-        _storage: &mut S,
-        _network: &mut N,
+        storage: &mut S,
+        network: &mut N,
     ) -> SyncResult<bool> {
-        tracing::warn!(
-            "⚠️ Headers2 support is currently NON-FUNCTIONAL. Received {} compressed headers from peer {} but cannot process them.",
+        tracing::info!(
+            "📦 Received {} compressed headers from peer {}",
             headers2.headers.len(),
             peer_id
         );
 
-        // Mark headers2 as failed for this session to avoid retrying
-        self.headers2_failed = true;
-
-        // Return an error to trigger fallback to regular headers
-        return Err(SyncError::Headers2DecompressionFailed(
-            "Headers2 is currently disabled due to protocol compatibility issues".to_string(),
-        ));
-
-        #[allow(unreachable_code)]
-        {
-            // If this is the first headers2 message, and we need to initialize compression state
-            if !headers2.headers.is_empty() {
-                // Check if we need to initialize the compression state
-                let state = self.headers2_state.get_state(peer_id);
-                if state.prev_header.is_none() {
-                    // If we're syncing from genesis (height 0), initialize with genesis header
-                    if self.chain_state.read().await.tip_height() == 0 {
-                        // We have genesis header at index 0
-                        if let Some(genesis_header) =
-                            self.chain_state.read().await.header_at_height(0)
-                        {
-                            tracing::info!(
+        // If this is the first headers2 message, and we need to initialize compression state
+        if !headers2.headers.is_empty() {
+            // Check if we need to initialize the compression state
+            let state = self.headers2_state.get_state(peer_id);
+            if state.prev_header.is_none() {
+                // If we're syncing from genesis (height 0), initialize with genesis header
+                if self.chain_state.read().await.tip_height() == 0 {
+                    // We have genesis header at index 0
+                    if let Some(genesis_header) = self.chain_state.read().await.header_at_height(0)
+                    {
+                        tracing::info!(
                             "Initializing headers2 compression state for peer {} with genesis header",
                             peer_id
                         );
-                            self.headers2_state.init_peer_state(peer_id, *genesis_header);
-                        }
-                    } else if self.chain_state.read().await.tip_height() > 0 {
-                        // Get our current tip to use as the base for compression
-                        if let Some(tip_header) = self.chain_state.read().await.get_tip_header() {
-                            tracing::info!(
+                        self.headers2_state.init_peer_state(peer_id, *genesis_header);
+                    }
+                } else if self.chain_state.read().await.tip_height() > 0 {
+                    // Get our current tip to use as the base for compression
+                    if let Some(tip_header) = self.chain_state.read().await.get_tip_header() {
+                        tracing::info!(
                             "Initializing headers2 compression state for peer {} with tip header at height {}",
                             peer_id,
                             self.chain_state.read().await.tip_height()
                         );
-                            self.headers2_state.init_peer_state(peer_id, tip_header);
-                        }
+                        self.headers2_state.init_peer_state(peer_id, tip_header);
                     }
                 }
             }
+        }
 
-            // Decompress headers using the peer's compression state
-            let headers = match self
-                .headers2_state
-                .process_headers(peer_id, headers2.headers.clone())
-            {
-                Ok(headers) => headers,
-                Err(e) => {
-                    tracing::error!(
+        // Decompress headers using the peer's compression state
+        let headers = match self.headers2_state.process_headers(peer_id, headers2.headers.clone()) {
+            Ok(headers) => headers,
+            Err(e) => {
+                tracing::error!(
                     "Failed to decompress headers2 from peer {}: {}. Headers count: {}, first header compressed: {}, chain height: {}",
                     peer_id,
                     e,
@@ -605,37 +582,36 @@ impl<S: StorageManager + Send + Sync + 'static, N: NetworkManager + Send + Sync
                     self.chain_state.read().await.tip_height()
                 );
 
-                    // If we failed due to missing previous header, and we're at genesis,
-                    // this might be a protocol issue where peer expects us to have genesis in compression state
-                    if matches!(e, crate::sync::headers2::ProcessError::DecompressionError(0, _))
-                        && self.chain_state.read().await.tip_height() == 0
-                    {
-                        tracing::warn!(
+                // If we failed due to missing previous header, and we're at genesis,
+                // this might be a protocol issue where peer expects us to have genesis in compression state
+                if matches!(e, crate::sync::headers2::ProcessError::DecompressionError(0, _))
+                    && self.chain_state.read().await.tip_height() == 0
+                {
+                    tracing::warn!(
                         "Headers2 decompression failed at genesis. Peer may be sending compressed headers that reference genesis. Consider falling back to regular headers."
                     );
-                    }
-
-                    // Return a specific error that can trigger fallback
-                    // Mark that headers2 failed for this sync session
-                    self.headers2_failed = true;
-                    return Err(SyncError::Headers2DecompressionFailed(format!(
-                        "Failed to decompress headers: {}",
-                        e
-                    )));
                 }
-            };
 
-            // Log compression statistics
-            let stats = self.headers2_state.get_stats();
-            tracing::info!(
-                "📊 Headers2 compression stats: {:.1}% bandwidth saved, {:.1}% compression ratio",
-                stats.bandwidth_savings,
-                stats.compression_ratio * 100.0
-            );
+                // Return a specific error that can trigger fallback
+                // Mark that headers2 failed for this sync session
+                self.headers2_failed = true;
+                return Err(SyncError::Headers2DecompressionFailed(format!(
+                    "Failed to decompress headers: {}",
+                    e
+                )));
+            }
+        };
 
-            // Process decompressed headers through the normal flow
-            self.handle_headers_message(headers, _storage, _network).await
-        }
+        // Log compression statistics
+        let stats = self.headers2_state.get_stats();
+        tracing::info!(
+            "📊 Headers2 compression stats: {:.1}% bandwidth saved, {:.1}% compression ratio",
+            stats.bandwidth_savings,
+            stats.compression_ratio * 100.0
+        );
+
+        // Process decompressed headers through the normal flow
+        self.handle_headers_message(headers, storage, network).await
     }
 
     /// Prepare sync state without sending network requests.
diff --git a/dash-spv/src/sync/message_handlers.rs b/dash-spv/src/sync/message_handlers.rs
@@ -312,7 +312,7 @@ impl<
         // Update phase state and check if we need to transition
         let should_transition = if let SyncPhase::DownloadingHeaders {
             current_height,
-
+            received_empty_response,
             last_progress,
             ..
         } = &mut self.current_phase
@@ -323,11 +323,16 @@ impl<
             // Note: We can't easily track headers_downloaded for compressed headers
             // without decompressing first, so we rely on the header sync manager's internal stats
 
+            // Mark sync complete if continue_sync is false (no more headers)
+            if !continue_sync {
+                *received_empty_response = true;
+            }
+
             // Update progress time
             *last_progress = Instant::now();
 
             // Check if phase is complete
-            !continue_sync
+            !continue_sync || *received_empty_response
         } else {
             false
         };
diff --git a/dash/src/network/message_headers2.rs b/dash/src/network/message_headers2.rs
diff --git a/dash/tests/headers2_compatibility_test.rs b/dash/tests/headers2_compatibility_test.rs
