Week 3 Day 19: Implement HEAD-based sync and missing message retrieval with causal ordering

Author: Anbu-00001

backend/p2p-node/discovery.go

@@ -38,13 +38,15 @@ func (n *discoveryNotifee) HandlePeerFound(pi peer.AddrInfo) {
 	} else {
 		log.Printf("[Discovery] Peer connected: %s (addrs: %v)", pi.ID.String(), pi.Addrs)
 
-		// Broadcast sync_request to the network
+		// Broadcast head_exchange to the network
 		syncReq := NetworkEnvelope{
-			MsgType: "sync_request",
-			Payload: map[string]interface{}{},
+			MsgType: "head_exchange",
+			Payload: map[string]interface{}{
+				"heads": n.psm.gossipLog.Heads(),
+			},
 		}
 
-		log.Printf("[Discovery] SYNC_REQUEST_SENT: Requesting state sync after connecting to peer %s", pi.ID.String())
+		log.Printf("[Discovery] HEADS_SENT: Requesting state sync after connecting to peer %s", pi.ID.String())
 		if err := n.psm.Broadcast(syncReq); err != nil {
 			log.Printf("[Discovery] Failed to broadcast sync_request: %v", err)
 		}

mobile_app/lib/services/gossip_log_service.dart

@@ -42,6 +42,29 @@ class GossipLogService {
   /// Snapshot of current HEAD message IDs (unmodifiable copy).
   Set<String> get heads => Set.unmodifiable(_heads);
 
+  /// Returns a snapshot of the current HEAD message IDs (messages with no children)
+  List<String> getHeads() => _heads.toList();
+
+  /// Returns peer heads that are missing locally
+  List<String> findMissingMessages(List<String> peerHeads) {
+    return peerHeads.where((h) => !_log.containsKey(h) && !_pending.containsKey(h)).toList();
+  }
+
+  /// Fetches the requested messages and their ancestors if needed, sorted topologically.
+  List<NetworkEnvelope> fetchAndSortMessages(List<String> requestedIds) {
+    final result = <NetworkEnvelope>[];
+    for (final id in requestedIds) {
+      if (_log.containsKey(id)) {
+        result.add(_log[id]!);
+      } else if (_pending.containsKey(id)) {
+        result.add(_pending[id]!);
+      }
+    }
+    // Topological sort by Lamport clock
+    result.sort((a, b) => a.clock.compareTo(b.clock));
+    return result;
+  }
+
   /// Number of messages in the log (applied).
   int get logSize => _log.length;
 
@@ -55,7 +78,9 @@ class GossipLogService {
   ///
   /// Updates the Lamport clock using max(local, received)+1, then either
   /// applies the message immediately (if deps satisfied) or queues it pending.
-  void receive(NetworkEnvelope env) {
+  /// Day-19: Returns a list of missing dependency IDs, if any.
+  List<String> receive(NetworkEnvelope env) {
+    final missingDeps = <String>[];
     // Lamport clock update: max(local, received) + 1
     if (env.clock > _clock) {
       _clock = env.clock;
@@ -68,7 +93,7 @@ class GossipLogService {
 
     // Skip if already known (extra safety beyond P2PService dedup)
     if (_log.containsKey(env.msgId) || _pending.containsKey(env.msgId)) {
-      return;
+      return missingDeps;
     }
 
     if (_canApply(env)) {
@@ -78,7 +103,13 @@ class GossipLogService {
       _pending[env.msgId] = env;
       debugPrint(
           '[GossipLog] MESSAGE_PENDING: msg_id=${env.msgId} waiting for deps=${env.prevMsgIds}');
+      for (final depId in env.prevMsgIds) {
+        if (depId.isNotEmpty && !_log.containsKey(depId) && !_pending.containsKey(depId)) {
+          missingDeps.add(depId);
+        }
+      }
     }
+    return missingDeps;
   }
 
   /// Records a locally-originated (sent) message into the log without emitting

mobile_app/lib/services/p2p_service.dart

@@ -144,6 +144,19 @@ class P2PService {
       // Forward the raw envelope for any listeners that want it
       _envelopeStreamController.add(envelope);
 
+      if (envelope.msgType == 'head_exchange') {
+        _handleHeadExchange(envelope);
+        return;
+      }
+      if (envelope.msgType == 'message_request') {
+        _handleMessageRequest(envelope);
+        return;
+      }
+      if (envelope.msgType == 'message_response') {
+        _handleMessageResponse(envelope);
+        return;
+      }
+
       // Day-18: sync_request and sync_response bypass GossipLog.
       // They are bulk state transfer meta-messages, not causal operations.
       if (envelope.msgType == 'sync_request') {
@@ -157,12 +170,115 @@ class P2PService {
 
       // All other message types go through GossipLog for causal ordering.
       // _gossipLogSub will call _routeValidEnvelope when deps are satisfied.
-      _gossipLog.receive(envelope);
+      final missingDeps = _gossipLog.receive(envelope);
+      if (missingDeps.isNotEmpty) {
+        debugPrint('[P2P] MISSING_DETECTED: requesting missing deps $missingDeps');
+        _sendMessageRequest(missingDeps);
+      }
     } catch (e) {
       debugPrint('[P2P] Failed to parse incoming message: $e');
     }
   }
 
+  void _handleHeadExchange(NetworkEnvelope envelope) {
+    debugPrint('[P2P] HEADS_RECEIVED: from ${envelope.originPeer}');
+    final peerHeads = (envelope.payload['heads'] as List<dynamic>? ?? [])
+        .map((e) => e.toString())
+        .toList();
+
+    final missing = _gossipLog.findMissingMessages(peerHeads);
+    if (missing.isNotEmpty) {
+      debugPrint('[P2P] MISSING_DETECTED: requesting ${missing.length} messages');
+      _sendMessageRequest(missing);
+    }
+  }
+
+  void _handleMessageRequest(NetworkEnvelope envelope) {
+    final requestedIds = (envelope.payload['requested_ids'] as List<dynamic>? ?? [])
+        .map((e) => e.toString())
+        .toList();
+    
+    debugPrint('[P2P] MESSAGE_REQUEST_RECEIVED: for ${requestedIds.length} messages');
+
+    final messagesToSend = _gossipLog.fetchAndSortMessages(requestedIds);
+    if (messagesToSend.isNotEmpty) {
+      _sendMessageResponse(messagesToSend);
+    }
+  }
+
+  void _handleMessageResponse(NetworkEnvelope envelope) {
+    debugPrint('[P2P] MESSAGE_RESPONSE_RECEIVED: from ${envelope.originPeer}');
+    final messagesList = envelope.payload['messages'] as List<dynamic>? ?? [];
+
+    for (final msgData in messagesList) {
+      try {
+        final Map<String, dynamic> msgMap =
+            msgData is Map<String, dynamic>
+                ? msgData
+                : Map<String, dynamic>.from(msgData as Map);
+
+        final msgEnvelope = NetworkEnvelope.fromJson(msgMap);
+        
+        // Add to deduplication cache to prevent re-processing
+        if (_messageCache.isDuplicate(msgEnvelope.msgId)) {
+          continue;
+        }
+
+        // Process through gossip log
+        final missingDeps = _gossipLog.receive(msgEnvelope);
+        
+        if (missingDeps.isNotEmpty) {
+          debugPrint('[P2P] MISSING_DETECTED: requesting missing deps $missingDeps');
+          _sendMessageRequest(missingDeps);
+        }
+      } catch (e) {
+        debugPrint('[P2P] Failed to parse message in response: $e');
+      }
+    }
+  }
+
+  Future<void> _sendMessageRequest(List<String> requestedIds) async {
+    if (requestedIds.isEmpty) return;
+
+    final envelope = NetworkEnvelope(
+      msgId: 'msg_req_${DateTime.now().millisecondsSinceEpoch}',
+      msgType: 'message_request',
+      originPeer: '',
+      timestamp: DateTime.now().millisecondsSinceEpoch ~/ 1000,
+      payload: {
+        'requested_ids': requestedIds,
+      },
+    );
+
+    debugPrint('[P2P] MESSAGE_REQUEST_SENT: requesting ids $requestedIds');
+    
+    // Add to our own dedup cache to prevent self-echo
+    _messageCache.isDuplicate(envelope.msgId);
+    
+    await _sendEnvelopeHttp(envelope);
+  }
+
+  Future<void> _sendMessageResponse(List<NetworkEnvelope> messages) async {
+    final payloadMessages = messages.map((m) => m.toJson()).toList();
+
+    final envelope = NetworkEnvelope(
+      msgId: 'msg_resp_${DateTime.now().millisecondsSinceEpoch}',
+      msgType: 'message_response',
+      originPeer: '',
+      timestamp: DateTime.now().millisecondsSinceEpoch ~/ 1000,
+      payload: {
+        'messages': payloadMessages,
+      },
+    );
+
+    debugPrint('[P2P] MESSAGE_RESPONSE_SENT: sending ${messages.length} messages');
+    
+    // Add to our own dedup cache to prevent self-echo
+    _messageCache.isDuplicate(envelope.msgId);
+    
+    await _sendEnvelopeHttp(envelope);
+  }
+
   /// Routes a causally-validated envelope to the appropriate downstream handler.
   ///
   /// Called by the GossipLog subscription once all dependencies are satisfied.

mobile_app/test/gossip_log_service_test.dart

@@ -162,5 +162,69 @@ void main() {
       // clock = max(6, 3) + 1 = 7
       expect(service.clock, 7);
     });
+
+    test('Day 19: findMissingMessages correctly identifies unknown peer heads', () async {
+      final m1 = _makeEnvelope(msgId: 'msg_1', clock: 1);
+      service.receive(m1);
+      
+      final missing = service.findMissingMessages(['msg_3', 'msg_1']);
+      expect(missing, ['msg_3']);
+    });
+
+    test('Day 19: receive returns missing dependencies', () async {
+      final m3 = _makeEnvelope(msgId: 'msg_3', clock: 3, prevMsgIds: ['msg_2']);
+      
+      final missingDeps = service.receive(m3);
+      expect(missingDeps, ['msg_2']);
+      expect(service.pendingSize, 1);
+    });
+
+    test('Day 19: fetchAndSortMessages returns topologically sorted messages', () async {
+      final m1 = _makeEnvelope(msgId: 'msg_1', clock: 1);
+      final m2 = _makeEnvelope(msgId: 'msg_2', clock: 2, prevMsgIds: ['msg_1']);
+      final m3 = _makeEnvelope(msgId: 'msg_3', clock: 3, prevMsgIds: ['msg_2']);
+
+      service.receive(m1);
+      service.receive(m2);
+      service.receive(m3);
+
+      // Request in reverse order to ensure sorting works
+      final fetched = service.fetchAndSortMessages(['msg_3', 'msg_2']);
+      expect(fetched.length, 2);
+      expect(fetched[0].msgId, 'msg_2', reason: 'm2 has lower clock, should be first');
+      expect(fetched[1].msgId, 'msg_3', reason: 'm3 has higher clock, should be second');
+    });
+
+    test('Day 19 Integration: Missing message retrieval in causal order', () async {
+      // B starts with M1
+      final m1 = _makeEnvelope(msgId: 'm1', clock: 1);
+      service.receive(m1);
+      await pumpEventQueue();
+      expect(received.length, 1);
+
+      // B detects missing M3 (from head_exchange)
+      final missingFromHeads = service.findMissingMessages(['m3']);
+      expect(missingFromHeads, ['m3']);
+
+      // B receives M3, depends on M2
+      final m3 = _makeEnvelope(msgId: 'm3', clock: 3, prevMsgIds: ['m2']);
+      final missingDeps = service.receive(m3);
+      
+      expect(missingDeps, ['m2']); // B detects missing M2
+      await pumpEventQueue();
+      expect(received.length, 1, reason: 'm3 should be pending');
+
+      // B requests and receives M2
+      final m2 = _makeEnvelope(msgId: 'm2', clock: 2, prevMsgIds: ['m1']);
+      final missingDeps2 = service.receive(m2);
+
+      expect(missingDeps2, isEmpty); // M1 is known
+      await pumpEventQueue();
+
+      // M2 unblocks M3
+      expect(received.length, 3);
+      expect(received[1].msgId, 'm2');
+      expect(received[2].msgId, 'm3');
+    });
   });
 }