event_graph: header sync with tips, retrieves events header that are not ancestors of the tips

src/event_graph/mod.rs

@@ -595,19 +595,22 @@ impl EventGraph {
         }
 
         if missing_parents.is_empty() {
-            info!(target: "event_graph::dag_sync()", "[EVENTGRAPH] DAG synced successfully!");
+            info!(target: "event_graph::dag_sync", "[EVENTGRAPH] DAG synced successfully!");
             return Ok(())
         }
-
-        // Header sync first
-        // TODO: requesting headers should be in a way that we wouldn't
-        // recieve the same header(s) again, by sending our tip, other
-        // nodes should send back the ones after it
         let hdr_tree_name = format!("headers_{dag_name}");
         let header_dag = self.dag_store.read().await.get_dag(&hdr_tree_name);
+        let dag_timestamp = u64::from_str(&dag_name)?;
+        let our_tips =
+            self.dag_store.read().await.header_dags.get(&dag_timestamp).unwrap().1.clone();
         let mut headers_requests = FuturesUnordered::new();
         for channel in channels.iter() {
-            headers_requests.push(request_header(&channel, dag_name.clone(), comms_timeout))
+            headers_requests.push(request_header(
+                channel,
+                dag_name.clone(),
+                our_tips.clone(),
+                comms_timeout,
+            ))
         }
 
         while let Some(peer_headers) = headers_requests.next().await {
@@ -616,12 +619,12 @@ impl EventGraph {
 
         // start download payload
         if !fast_mode {
-            info!(target: "event_graph::dag_sync()", "[EVENTGRAPH] Fetching events");
+            info!(target: "event_graph::dag_sync", "[EVENTGRAPH] Fetching events");
             let mut header_sorted = vec![];
 
             let main_dag = self.dag_store.read().await.get_dag(&dag_name);
             for iter_elem in header_dag.iter() {
-                let (hash_bytes , val) = iter_elem.unwrap();
+                let (hash_bytes, val) = iter_elem.unwrap();
                 let val: Header = deserialize_async(&val).await.unwrap();
                 if val.parents != NULL_PARENTS && !main_dag.contains_key(hash_bytes)? {
                     header_sorted.push(val);
@@ -629,7 +632,7 @@ impl EventGraph {
             }
             header_sorted.sort_by(|x, y| x.layer.cmp(&y.layer));
 
-            info!(target: "event_graph::dag_sync()", "[EVENTGRAPH] Retrieving {} Events", header_sorted.len());
+            info!(target: "event_graph::dag_sync", "[EVENTGRAPH] Retrieving {} Events", header_sorted.len());
             // Implement parallel download of events with a batch size
             let batch = 20;
             // Mapping of the chunk group id to the chunk, using a BTreeMap help us to
@@ -689,7 +692,7 @@ impl EventGraph {
                     peer_status.insert(free_channels[i].address().clone(), PeerStatus::Busy);
                 }
 
-                info!(target: "event_graph::dag_sync()", "[EVENTGRAPH] Retrieving Events from {} peers", futures.len());
+                info!(target: "event_graph::dag_sync", "[EVENTGRAPH] Retrieving Events from {} peers", futures.len());
                 if let Some(resp) = futures.next().await {
                     let (events, chunk_id, channel) = resp;
                     if let Ok(events) = events {
@@ -701,7 +704,7 @@ impl EventGraph {
                         peer_status.insert(channel.address().clone(), PeerStatus::Failed);
                     }
 
-                    info!(target: "event_graph::dag_sync()", "[EVENTGRAPH] Retrieved Events: {}/{}", retrieved_count, header_sorted.len());
+                    info!(target: "event_graph::dag_sync", "[EVENTGRAPH] Retrieved Events: {}/{}", retrieved_count, header_sorted.len());
                 }
             }
 
@@ -709,12 +712,12 @@ impl EventGraph {
             for (_, chunk) in received_events {
                 verified_count += chunk.len();
                 self.dag_insert(&chunk, &dag_name).await?;
-                info!(target: "event_graph::dag_sync()", "[EVENTGRAPH] Verified Events: {}/{}", verified_count, retrieved_count);
+                info!(target: "event_graph::dag_sync", "[EVENTGRAPH] Verified Events: {}/{}", verified_count, retrieved_count);
             }
         }
         // <-- end download payload
 
-        info!(target: "event_graph::dag_sync()", "[EVENTGRAPH] DAG synced successfully!");
+        info!(target: "event_graph::dag_sync", "[EVENTGRAPH] DAG synced successfully!");
         Ok(())
     }
 
@@ -843,7 +846,7 @@ impl EventGraph {
             }
 
             if !event.dag_validate(&header_dag).await? {
-                error!(target: "event_graph::dag_insert()", "Event {} is invalid!", event_id);
+                error!(target: "event_graph::dag_insert", "Event {} is invalid!", event_id);
                 return Err(Error::EventIsInvalid)
             }
 
@@ -953,11 +956,11 @@ impl EventGraph {
                 continue
             }
             debug!(
-                target: "event_graph::header_dag_insert()",
+                target: "event_graph::header_dag_insert",
                 "Inserting header {} into the DAG", header_id,
             );
             if !header.validate(&header_dag, self.hours_rotation, Some(&overlay)).await? {
-                error!(target: "event_graph::header_dag_insert()", "Header {} is invalid!", header_id);
+                error!(target: "event_graph::header_dag_insert", "Header {} is invalid!", header_id);
                 return Err(Error::HeaderIsInvalid)
             }
             let header_se = serialize_async(&header).await;
@@ -1207,6 +1210,74 @@ impl EventGraph {
         JsonResponse::new(result, id).into()
     }
 
+    /// Fetch all events that are not ancestors of the tips
+    pub async fn fetch_headers_with_tips(
+        &self,
+        dag_name: &str,
+        tips: &LayerUTips,
+    ) -> Result<Vec<Header>> {
+        debug!(
+             target: "event_graph::fetch_headers_with_tips",
+             "fetching headers with tips {tips:?}"
+        );
+
+        let tree = self.dag_store.read().await.get_dag(&format!("headers_{dag_name}"));
+
+        // Let's identify all the events that are ancestors of the tips so that we don't send those
+        let mut ancestors = HashSet::new();
+
+        for hashes in tips.values() {
+            for hash in hashes {
+                ancestors.insert(*hash);
+                let val = tree
+                    .get(hash.as_bytes())?
+                    .ok_or_else(|| Error::EventNotFound("The Tip is not found".to_owned()))?;
+                let header: Header = deserialize_async(&val).await?;
+                self.get_ancestors(&mut ancestors, header, &tree).await?
+            }
+        }
+
+        let mut result = Vec::with_capacity(tree.len() - ancestors.len());
+        // Now find the events that are not ancestors of the tips
+        // it is a set difference operation: unseen_events = all_events - tip_ancestors
+        for iter_elem in tree.iter() {
+            let (id, val) = iter_elem?;
+            let hash = Hash::from_bytes((&id as &[u8]).try_into()?);
+            if !ancestors.contains(&hash) {
+                let header: Header = deserialize_async(&val).await?;
+                result.push(header);
+            }
+        }
+
+        result.sort_unstable_by(|a, b| a.layer.cmp(&b.layer));
+
+        Ok(result)
+    }
+
+    /// Finds all the ancestors of an event
+    async fn get_ancestors(
+        &self,
+        visited: &mut HashSet<Hash>,
+        header: Header,
+        tree: &sled::Tree,
+    ) -> Result<()> {
+        let mut stack = VecDeque::new();
+        stack.push_back(header);
+
+        while let Some(hdr) = stack.pop_back() {
+            for parent in hdr.parents {
+                if parent != NULL_ID && !visited.contains(&parent) {
+                    visited.insert(parent);
+                    let val = tree.get(parent.as_bytes())?.unwrap();
+                    let header: Header = deserialize_async(&val).await?;
+                    stack.push_back(header);
+                }
+            }
+        }
+
+        Ok(())
+    }
+
     /// Fetch all the events that are on a higher layers than the
     /// provided ones.
     pub async fn fetch_successors_of(&self, tips: LayerUTips) -> Result<Vec<Event>> {
@@ -1218,7 +1289,7 @@ impl EventGraph {
         let current_genesis = self.current_genesis.read().await;
         let dag_name = current_genesis.header.timestamp.to_string();
         let mut graph = HashMap::new();
-        for iter_elem in self.dag_store.read().await.get_dag(dag_name).iter() {
+        for iter_elem in self.dag_store.read().await.get_dag(&dag_name).iter() {
             let (id, val) = iter_elem.unwrap();
             let hash = Hash::from_bytes((&id as &[u8]).try_into().unwrap());
             let event: Event = deserialize_async(&val).await.unwrap();
@@ -1367,6 +1438,7 @@ impl EventGraph {
 async fn request_header(
     peer: &Channel,
     tree_name: String,
+    tips: LayerUTips,
     comms_timeout: u64,
 ) -> Result<Vec<Header>> {
     let url = peer.address();
@@ -1375,7 +1447,7 @@ async fn request_header(
         Ok(v) => v,
         Err(e) => {
             error!(
-                target: "event_graph::dag_sync()",
+                target: "event_graph::dag_sync",
                 "[EVENTGRAPH] Sync: Couldn't subscribe HeaderReq for peer {}, skipping ({})",
                 url, e,
             );
@@ -1383,9 +1455,9 @@ async fn request_header(
         }
     };
 
-    if let Err(e) = peer.send(&HeaderReq(tree_name)).await {
+    if let Err(e) = peer.send(&HeaderReq(tree_name, tips)).await {
         error!(
-            target: "event_graph::dag_sync()",
+            target: "event_graph::dag_sync",
             "[EVENTGRAPH] Sync: Couldn't contact peer {}, skipping ({})", url, e,
         );
         return Err(Error::EventNotFound("Couldn't contact peer".to_owned()));
@@ -1394,7 +1466,7 @@ async fn request_header(
     // Node waits for response
     let Ok(peer_headers) = hdr_rep_sub.receive_with_timeout(comms_timeout).await else {
         error!(
-            target: "event_graph::dag_sync()",
+            target: "event_graph::dag_sync",
             "[EVENTGRAPH] Sync: Peer {} didn't reply with headers in time, skipping", url,
         );
         // communicated_peers -= 1;
@@ -1415,7 +1487,7 @@ async fn request_event(
     let url = peer.address();
 
     debug!(
-        target: "event_graph::dag_sync()",
+        target: "event_graph::dag_sync",
         "Requesting {:?} from {}...", headers, url,
     );
 
@@ -1423,7 +1495,7 @@ async fn request_event(
         Ok(v) => v,
         Err(e) => {
             error!(
-                target: "event_graph::dag_sync()",
+                target: "event_graph::dag_sync",
                 "[EVENTGRAPH] Sync: Couldn't subscribe EventRep for peer {}, skipping ({})",
                 url, e,
             );
@@ -1438,7 +1510,7 @@ async fn request_event(
     // let request_missing_events = missing_parents.clone().into_iter().collect();
     if let Err(e) = peer.send(&EventReq(headers.clone())).await {
         error!(
-            target: "event_graph::dag_sync()",
+            target: "event_graph::dag_sync",
             "[EVENTGRAPH] Sync: Failed communicating EventReq({:?}) to {}: {}",
             headers, url, e,
         );
@@ -1452,7 +1524,7 @@ async fn request_event(
     // Node waits for response
     let Ok(event) = ev_rep_sub.receive_with_timeout(comms_timeout).await else {
         error!(
-            target: "event_graph::dag_sync()",
+            target: "event_graph::dag_sync",
             "[EVENTGRAPH] Sync: Timeout waiting for parents {:?} from {}",
             headers, url,
         );

src/event_graph/proto.rs

@@ -174,7 +174,7 @@ impl_p2p_message!(HeaderPut, "EventGraph::HeaderPut", 0, 0, DEFAULT_METERING_CON
 
 /// A P2P message representing a header request
 #[derive(Clone, SerialEncodable, SerialDecodable)]
-pub struct HeaderReq(pub String);
+pub struct HeaderReq(pub String, pub LayerUTips);
 impl_p2p_message!(HeaderReq, "EventGraph::HeaderReq", 0, 0, DEFAULT_METERING_CONFIGURATION);
 
 /// A P2P message representing a header reply
@@ -917,10 +917,9 @@ impl ProtocolEventGraph {
     /// sending their current headers.
     async fn handle_header_req(self: Arc<Self>) -> Result<()> {
         loop {
-            let dag_name = match self.hdr_req_sub.receive().await {
-                Ok(v) => v.0.clone(),
-                Err(_) => continue,
-            };
+            let Ok(v) = self.hdr_req_sub.receive().await else { continue };
+            let (dag_name, tips) = (&v.0, &v.1);
+
             trace!(
                 target: "event_graph::protocol::handle_tip_req",
                 "Got TipReq [{}]", self.channel.display_address(),
@@ -939,20 +938,12 @@ impl ProtocolEventGraph {
 
             // We received header request. Let's find them, add them to
             // our bcast ids list, and reply with them.
-            let dag_timestamp = u64::from_str(&dag_name)?;
+            let dag_timestamp = u64::from_str(dag_name)?;
             let store = self.event_graph.dag_store.read().await;
             if !store.header_dags.contains_key(&dag_timestamp) {
                 continue
             }
-            let main_dag = store.get_dag(&dag_name);
-            let mut headers = vec![];
-            for item in main_dag.iter() {
-                let (_, event) = item.unwrap();
-                let event: Event = deserialize_async(&event).await.unwrap();
-                if !headers.contains(&event.header) || event.header.parents != NULL_PARENTS {
-                    headers.push(event.header);
-                }
-            }
+            let headers = self.event_graph.fetch_headers_with_tips(dag_name, tips).await?;
             // let mut bcast_ids = self.event_graph.broadcasted_ids.write().await;
             // for (_, tips) in layers.iter() {
             //     for tip in tips {

src/event_graph/tests.rs

@@ -881,3 +881,459 @@ fn dag_insert_valid_event() -> Result<()> {
         Ok(())
     })
 }
+
+/*
+   This function builds the following graph
+
+   Layer    3           2                    1                    0
+        [Event3A]-----[Event2A]-------|
+                                      |-----[Event1A]-----|
+                                               |          |
+                                      ---------|          |
+        [Event3B]-----[Event2B]-------|                   |
+                                      |                   |
+                                      |-----[Event1B]-----|-----[GENESIS]
+                                                          |
+        [Event3C]-----[Event2C]----|                      |
+                                   |  |-----[Event1C]-----|
+                                   ---|                   |
+                                   |  |                   |
+        [Event3D]-----[Event2D]----|  |------[Event1D]----|
+*/
+async fn build_graph() -> Result<(EventGraphPtr, Vec<Event>)> {
+    let event_graph = make_event_graph().await?;
+    let mut events = vec![];
+    let dag_name = event_graph
+        .dag_store
+        .read()
+        .await
+        .main_dags
+        .last_key_value()
+        .unwrap()
+        .0
+        .clone()
+        .to_string();
+
+    let current_dag_genesis_hash = event_graph.current_genesis.read().await.id();
+
+    // first layer
+    let mut parents = [NULL_ID; N_EVENT_PARENTS];
+    parents[0] = current_dag_genesis_hash;
+    let event1a = Event {
+        header: Header {
+            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64 + 1,
+            layer: 1,
+            parents,
+        },
+        content: "Event1A".as_bytes().to_vec(),
+    };
+    events.push(event1a.clone());
+
+    let event1b = Event {
+        header: Header {
+            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64 + 2,
+            layer: 1,
+            parents,
+        },
+        content: "Event1B".as_bytes().to_vec(),
+    };
+    events.push(event1b.clone());
+
+    let event1c = Event {
+        header: Header {
+            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64 + 3,
+            layer: 1,
+            parents,
+        },
+        content: "Event1C".as_bytes().to_vec(),
+    };
+    events.push(event1c.clone());
+
+    let event1d = Event {
+        header: Header {
+            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64 + 4,
+            layer: 1,
+            parents,
+        },
+        content: "Event1D".as_bytes().to_vec(),
+    };
+    events.push(event1d.clone());
+
+    // second layer
+    parents[0] = event1a.id();
+    let event2a = Event {
+        header: Header {
+            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64 + 5,
+            layer: 2,
+            parents,
+        },
+        content: "Event2A".as_bytes().to_vec(),
+    };
+    events.push(event2a.clone());
+
+    parents[1] = event1b.id();
+    let event2b = Event {
+        header: Header {
+            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64 + 6,
+            layer: 2,
+            parents,
+        },
+        content: "Event2B".as_bytes().to_vec(),
+    };
+    events.push(event2b.clone());
+
+    parents[0] = event1c.id();
+    parents[1] = event1d.id();
+    let event2c = Event {
+        header: Header {
+            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64 + 7,
+            layer: 2,
+            parents,
+        },
+        content: "Event2C".as_bytes().to_vec(),
+    };
+    events.push(event2c.clone());
+
+    let event2d = Event {
+        header: Header {
+            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64 + 8,
+            layer: 2,
+            parents,
+        },
+        content: "Event2D".as_bytes().to_vec(),
+    };
+    events.push(event2d.clone());
+
+    // third layer
+    let mut parents = [NULL_ID; N_EVENT_PARENTS];
+    parents[0] = event2a.id();
+    let event3a = Event {
+        header: Header {
+            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64 + 9,
+            layer: 3,
+            parents,
+        },
+        content: "Event3A".as_bytes().to_vec(),
+    };
+    events.push(event3a.clone());
+
+    parents[0] = event2b.id();
+    let event3b = Event {
+        header: Header {
+            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64 + 10,
+            layer: 3,
+            parents,
+        },
+        content: "Event3B".as_bytes().to_vec(),
+    };
+    events.push(event3b.clone());
+
+    parents[0] = event2c.id();
+    let event3c = Event {
+        header: Header {
+            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64 + 11,
+            layer: 3,
+            parents,
+        },
+        content: "Event3C".as_bytes().to_vec(),
+    };
+    events.push(event3c.clone());
+
+    parents[0] = event2d.id();
+    let event3d = Event {
+        header: Header {
+            timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64 + 12,
+            layer: 3,
+            parents,
+        },
+        content: "Event3D".as_bytes().to_vec(),
+    };
+    events.push(event3d.clone());
+
+    // Insert events 1a to 1d
+    event_graph
+        .header_dag_insert(
+            vec![
+                event1a.header.clone(),
+                event1b.header.clone(),
+                event1c.header.clone(),
+                event1d.header.clone(),
+            ],
+            &dag_name,
+        )
+        .await?;
+    event_graph.dag_insert(&[event1a, event1b, event1c, event1d], &dag_name).await?;
+    // Insert events 2a to 2d
+    event_graph
+        .header_dag_insert(
+            vec![
+                event2a.header.clone(),
+                event2b.header.clone(),
+                event2c.header.clone(),
+                event2d.header.clone(),
+            ],
+            &dag_name,
+        )
+        .await?;
+    event_graph.dag_insert(&[event2a, event2b, event2c, event2d], &dag_name).await?;
+    // Insert events 3a to 3d
+    event_graph
+        .header_dag_insert(
+            vec![
+                event3a.header.clone(),
+                event3b.header.clone(),
+                event3c.header.clone(),
+                event3d.header.clone(),
+            ],
+            &dag_name,
+        )
+        .await?;
+    event_graph.dag_insert(&[event3a, event3b, event3c, event3d], &dag_name).await?;
+
+    //panic!("REACHED HERE");
+
+    Ok((event_graph, events))
+}
+
+#[test]
+fn find_ancestors_of_an_event() -> Result<()> {
+    smol::block_on(async {
+        let (event_graph, events) = build_graph().await?;
+
+        let dag_name = event_graph
+            .dag_store
+            .read()
+            .await
+            .main_dags
+            .last_key_value()
+            .unwrap()
+            .0
+            .clone()
+            .to_string();
+
+        let tree = event_graph.dag_store.read().await.get_dag(&format!("headers_{dag_name}"));
+
+        let events_map: HashMap<String, Event> =
+            events.into_iter().map(|e| (String::from_utf8_lossy(&e.content).into(), e)).collect();
+        let genesis_header = event_graph.current_genesis.read().await.header.clone();
+        let genesis_hash = genesis_header.id();
+        // Genesis layer
+        let mut genesis_ancestors = HashSet::new();
+        event_graph.get_ancestors(&mut genesis_ancestors, genesis_header, &tree).await?;
+        assert!(genesis_ancestors.is_empty());
+
+        // 1st layer
+        let mut event1a_ancestors = HashSet::new();
+        event_graph
+            .get_ancestors(
+                &mut event1a_ancestors,
+                events_map.get("Event1A").unwrap().header.clone(),
+                &tree,
+            )
+            .await?;
+        let mut event1b_ancestors = HashSet::new();
+        event_graph
+            .get_ancestors(
+                &mut event1b_ancestors,
+                events_map.get("Event1B").unwrap().header.clone(),
+                &tree,
+            )
+            .await?;
+        let mut event1c_ancestors = HashSet::new();
+        event_graph
+            .get_ancestors(
+                &mut event1c_ancestors,
+                events_map.get("Event1C").unwrap().header.clone(),
+                &tree,
+            )
+            .await?;
+        let mut event1d_ancestors = HashSet::new();
+        event_graph
+            .get_ancestors(
+                &mut event1d_ancestors,
+                events_map.get("Event1D").unwrap().header.clone(),
+                &tree,
+            )
+            .await?;
+
+        // Only genesis is the ancestor
+        assert!(event1a_ancestors.len() == 1 && event1a_ancestors.contains(&genesis_hash));
+        assert!(event1b_ancestors.len() == 1 && event1b_ancestors.contains(&genesis_hash));
+        assert!(event1c_ancestors.len() == 1 && event1c_ancestors.contains(&genesis_hash));
+        assert!(event1d_ancestors.len() == 1 && event1d_ancestors.contains(&genesis_hash));
+
+        // 2nd layer
+        let event2a_expected_ancestors =
+            HashSet::from([genesis_hash, events_map.get("Event1A").unwrap().id()]);
+        let event2b_expected_ancestors = HashSet::from([
+            genesis_hash,
+            events_map.get("Event1B").unwrap().id(),
+            events_map.get("Event1A").unwrap().id(),
+        ]);
+        let event2cd_expected_ancestors = HashSet::from([
+            genesis_hash,
+            events_map.get("Event1C").unwrap().id(),
+            events_map.get("Event1D").unwrap().id(),
+        ]);
+
+        let mut event2a_ancestors = HashSet::new();
+        event_graph
+            .get_ancestors(
+                &mut event2a_ancestors,
+                events_map.get("Event2A").unwrap().header.clone(),
+                &tree,
+            )
+            .await?;
+        let mut event2b_ancestors = HashSet::new();
+        event_graph
+            .get_ancestors(
+                &mut event2b_ancestors,
+                events_map.get("Event2B").unwrap().header.clone(),
+                &tree,
+            )
+            .await?;
+        let mut event2c_ancestors = HashSet::new();
+        event_graph
+            .get_ancestors(
+                &mut event2c_ancestors,
+                events_map.get("Event2C").unwrap().header.clone(),
+                &tree,
+            )
+            .await?;
+        let mut event2d_ancestors = HashSet::new();
+        event_graph
+            .get_ancestors(
+                &mut event2d_ancestors,
+                events_map.get("Event2D").unwrap().header.clone(),
+                &tree,
+            )
+            .await?;
+
+        assert_eq!(event2a_ancestors, event2a_expected_ancestors);
+        assert_eq!(event2b_ancestors, event2b_expected_ancestors);
+        assert_eq!(event2c_ancestors, event2cd_expected_ancestors);
+        assert_eq!(event2d_ancestors, event2cd_expected_ancestors);
+
+        // 3rd layer
+        let mut event3a_expected_ancestors = event2a_expected_ancestors.clone();
+        event3a_expected_ancestors.insert(events_map.get("Event2A").unwrap().header.clone().id());
+        let mut event3b_expected_ancestors = event2b_expected_ancestors.clone();
+        event3b_expected_ancestors.insert(events_map.get("Event2B").unwrap().header.clone().id());
+        let mut event3c_expected_ancestors = event2cd_expected_ancestors.clone();
+        event3c_expected_ancestors.insert(events_map.get("Event2C").unwrap().header.clone().id());
+        let mut event3d_expected_ancestors = event2cd_expected_ancestors.clone();
+        event3d_expected_ancestors.insert(events_map.get("Event2D").unwrap().header.clone().id());
+
+        let mut event3a_ancestors = HashSet::new();
+        event_graph
+            .get_ancestors(
+                &mut event3a_ancestors,
+                events_map.get("Event3A").unwrap().header.clone(),
+                &tree,
+            )
+            .await?;
+        let mut event3b_ancestors = HashSet::new();
+        event_graph
+            .get_ancestors(
+                &mut event3b_ancestors,
+                events_map.get("Event3B").unwrap().header.clone(),
+                &tree,
+            )
+            .await?;
+        let mut event3c_ancestors = HashSet::new();
+        event_graph
+            .get_ancestors(
+                &mut event3c_ancestors,
+                events_map.get("Event3C").unwrap().header.clone(),
+                &tree,
+            )
+            .await?;
+        let mut event3d_ancestors = HashSet::new();
+        event_graph
+            .get_ancestors(
+                &mut event3d_ancestors,
+                events_map.get("Event3D").unwrap().header.clone(),
+                &tree,
+            )
+            .await?;
+
+        assert_eq!(event3a_ancestors, event3a_expected_ancestors);
+        assert_eq!(event3b_ancestors, event3b_expected_ancestors);
+        assert_eq!(event3c_ancestors, event3c_expected_ancestors);
+        assert_eq!(event3d_ancestors, event3d_expected_ancestors);
+
+        Ok(())
+    })
+}
+
+#[test]
+fn fetches_headers_with_tips() -> Result<()> {
+    smol::block_on(async {
+        let (event_graph, events) = build_graph().await?;
+
+        let dag_name = event_graph
+            .dag_store
+            .read()
+            .await
+            .main_dags
+            .last_key_value()
+            .unwrap()
+            .0
+            .clone()
+            .to_string();
+
+        let map: HashMap<blake3::Hash, String> = events
+            .into_iter()
+            .map(|e| (e.id(), String::from_utf8_lossy(&e.content).into()))
+            .collect();
+        let name_map: HashMap<String, blake3::Hash> =
+            map.iter().map(|(hash, content)| (content.clone(), *hash)).collect();
+
+        let genesis_hash = event_graph.current_genesis.read().await.id();
+        let patha = ["Event3A", "Event2A", "Event1A"];
+        let pathb = ["Event3B", "Event2B", "Event1B", "Event1A"];
+        let pathc = ["Event3C", "Event2C", "Event1C", "Event1D"];
+        let pathd = ["Event3D", "Event2D", "Event1C", "Event1D"];
+
+        let patha_tip = BTreeMap::from([(3, HashSet::from([*name_map.get("Event3A").unwrap()]))]);
+        // Should be only headers that are not ancestors of Event3A
+        let headers = event_graph.fetch_headers_with_tips(&dag_name, &patha_tip).await?;
+        assert!(headers.iter().all(
+            |h| h.id() != genesis_hash && !patha.contains(&map.get(&h.id()).unwrap().as_str())
+        ));
+
+        let pathb_tip = BTreeMap::from([(3, HashSet::from([*name_map.get("Event3B").unwrap()]))]);
+        // Should be only headers that are not ancestors of Event3B
+        let headers = event_graph.fetch_headers_with_tips(&dag_name, &pathb_tip).await?;
+        assert!(headers.iter().all(
+            |h| h.id() != genesis_hash && !pathb.contains(&map.get(&h.id()).unwrap().as_str())
+        ));
+
+        let pathc_tip = BTreeMap::from([(3, HashSet::from([*name_map.get("Event3C").unwrap()]))]);
+        // Should be only headers that are not ancestors of Event3C
+        let headers = event_graph.fetch_headers_with_tips(&dag_name, &pathc_tip).await?;
+        assert!(headers.iter().all(
+            |h| h.id() != genesis_hash && !pathc.contains(&map.get(&h.id()).unwrap().as_str())
+        ));
+
+        let pathd_tip = BTreeMap::from([(3, HashSet::from([*name_map.get("Event3D").unwrap()]))]);
+        // Should be only headers that are not ancestors of Event3D
+        let headers = event_graph.fetch_headers_with_tips(&dag_name, &pathd_tip).await?;
+        assert!(headers.iter().all(
+            |h| h.id() != genesis_hash && !pathd.contains(&map.get(&h.id()).unwrap().as_str())
+        ));
+
+        // Two tips Event3A and Event3D
+        let mut comb_tip = BTreeMap::new();
+        comb_tip.extend(patha_tip);
+        comb_tip.get_mut(&3).unwrap().extend(pathd_tip.get(&3).unwrap());
+
+        // Should be only headers that are not ancestors of Event3A and Event3D
+        let headers = event_graph.fetch_headers_with_tips(&dag_name, &comb_tip).await?;
+        assert!(headers.iter().all(|h| h.id() != genesis_hash &&
+            !patha.contains(&map.get(&h.id()).unwrap().as_str()) &&
+            !pathd.contains(&map.get(&h.id()).unwrap().as_str())));
+
+        Ok(())
+    })
+}