feat: local blacklists

aizensoosuke · aizensoosuke · commit 5be91c72e86c · 2025-07-03T23:17:55.000+02:00
diff --git a/Makefile b/Makefile
@@ -129,6 +129,17 @@ deploy:
 	cd cli && cargo build --release --target=x86_64-unknown-linux-musl -p fledger && cargo build --release --target=x86_64-unknown-linux-musl -p flsignal
 	@./deploy-binaries.sh
 
+test_realm:
+	/usr/bin/mktemp -d -t fledger-XXXX
+	./target-common/x86_64-unknown-linux-musl/release/fledger --config "$(shell /usr/bin/mktemp -d -t fledger-XXXX)" --disable-turn-stun --signal-url ws://localhost:8765 realm create simulation --cond-pass
+
+test_signal:
+	./target-common/x86_64-unknown-linux-musl/release/flsignal -v --max-list-len 25
+
+test_just_fetch_once:
+	rm -rf /tmp/fledger
+	./target-common/x86_64-unknown-linux-musl/release/fledger --config "$(shell /usr/bin/mktemp -d -t fledger-XXXX)" --disable-turn-stun --signal-url ws://localhost:8765 simulation just-fetch-once
+
 clean:
 	for c in ${CARGOS}; do \
 		echo "Cleaning $$c"; \
diff --git a/cli/fledger/src/realm.rs b/cli/fledger/src/realm.rs
@@ -49,6 +49,7 @@ impl RealmHandler {
         cond_pass: bool,
     ) -> anyhow::Result<()> {
         f.loop_node(crate::FledgerState::Connected(1)).await?;
+        //f.loop_node(crate::FledgerState::Duration(1)).await?;
 
         let config = RealmConfig {
             max_space: max_space.unwrap_or(1000000),
diff --git a/cli/fledger/src/simulation.rs b/cli/fledger/src/simulation.rs
@@ -63,7 +63,12 @@ pub enum SimulationSubcommand {
 
         #[arg(long)]
         experiment_id: u32,
+
+        #[arg(long, default_value = "false")]
+        with_local_blacklists: bool,
     },
+
+    JustFetchOnce {},
 }
 
 pub struct SimulationHandler {}
@@ -108,6 +113,7 @@ impl SimulationHandler {
                 timeout_ms,
                 enable_sync,
                 experiment_id,
+                with_local_blacklists,
             } => {
                 let evil_noforward = f.args.evil_noforward.clone();
                 SimulationDht::run_fetch_pages(
@@ -118,9 +124,11 @@ impl SimulationHandler {
                     timeout_ms,
                     experiment_id,
                     evil_noforward,
+                    with_local_blacklists,
                 )
                 .await
             }
+            SimulationSubcommand::JustFetchOnce {} => SimulationDht::just_fetch_once(f).await,
         }
     }
 }
diff --git a/cli/fledger/src/simulation_dht/simulation.rs b/cli/fledger/src/simulation_dht/simulation.rs
@@ -186,9 +186,18 @@ impl SimulationDht {
         timeout_ms: u32,
         experiment_id: u32,
         evil_noforward: bool,
+        with_local_blacklists: bool,
     ) -> anyhow::Result<()> {
         let mut start_instant = Instant::now();
 
+        unsafe {
+            if with_local_blacklists {
+                log::info!("enabling local blacklists");
+                flmodules::dht_storage::messages::LOCAL_BLACKLISTS = true;
+                flmodules::dht_router::messages::LOCAL_BLACKLISTS = true;
+            }
+        }
+
         let node_name = f.node.node_config.info.name.clone();
         let mut state = SimulationState::new(experiment_id, node_name);
 
@@ -301,4 +310,34 @@ impl SimulationDht {
             }
         }
     }
+
+    pub async fn just_fetch_once(mut f: Fledger) -> anyhow::Result<()> {
+        f.loop_node(crate::FledgerState::DHTAvailable).await?;
+        f.loop_node(crate::FledgerState::Connected(2)).await?;
+
+        let realm_id = RealmView::new_first(f.node.dht_storage.as_ref().unwrap().clone())
+            .await?
+            .realm
+            .realm_id();
+
+        let flo_id =
+            FloID::from_str("5efe0a6143df5641af9d6036ba8da82222bb30211c21ba5ec236851efda38420")?;
+        let global_page_id = Self::make_page_id(realm_id.clone(), flo_id);
+
+        for i in 0..10 {
+            wait_ms(1000).await;
+
+            let page = Self::fetch_page(&mut f, global_page_id.clone()).await;
+
+            if page.is_ok() {
+                log::info!("God made a miracle");
+            } else {
+                log::info!("Fetch {i} done.");
+            }
+        }
+
+        f.loop_node(crate::FledgerState::Forever).await?;
+
+        Ok(())
+    }
 }
diff --git a/flmodules/src/dht_router/messages.rs b/flmodules/src/dht_router/messages.rs
@@ -1,13 +1,17 @@
+use std::collections::{HashMap, HashSet};
+
 use flarch::{
     broker::{SubsystemHandler, TranslateFrom, TranslateInto},
     nodeids::{NodeID, U256},
     platform_async_trait,
 };
-use rand::seq::SliceRandom;
+use itertools::Itertools;
+use rand::{seq::SliceRandom, Rng};
 use serde::{Deserialize, Serialize};
 use tokio::sync::watch;
 
 use crate::{
+    dht_storage::messages::MessageClosest,
     nodeconfig::NodeInfo,
     router::messages::{NetworkWrapper, RouterIn, RouterOut},
     timer::TimerMessage,
@@ -19,6 +23,7 @@ use super::{
 };
 
 pub static mut EVIL_NO_FORWARD: bool = false;
+pub static mut LOCAL_BLACKLISTS: bool = false;
 
 /// These are the messages which will be exchanged between the nodes for this
 /// module.
@@ -73,6 +78,11 @@ pub(super) struct Messages {
     // This is different than core.active, because there can be connections from other
     // modules, or connections from another node.
     connected: Vec<NodeID>,
+
+    // readFlo id -> node id
+    requests_in_flight: HashMap<U256, U256>,
+
+    blacklisted_nodes: HashSet<U256>,
 }
 
 impl Messages {
@@ -84,6 +94,8 @@ impl Messages {
                 core: Kademlia::new(root, cfg),
                 tx: Some(tx),
                 connected: vec![],
+                requests_in_flight: HashMap::new(),
+                blacklisted_nodes: HashSet::new(),
             },
             rx,
         )
@@ -233,22 +245,131 @@ impl Messages {
         }
     }
 
+    fn get_readflo_id(&self, msg: NetworkWrapper) -> Option<U256> {
+        if msg.module == "DHTStorage" {
+            match msg.unwrap_yaml("DHTStorage") {
+                Some(msg_readflo) => match msg_readflo {
+                    MessageClosest::ReadFlo(_id, request_id) => {
+                        log::warn!("sending MessageClosest::ReadFlo {}", msg.msg.clone());
+                        Some(request_id.clone())
+                    }
+                    _ => None,
+                },
+                None => None,
+            }
+        } else {
+            None
+        }
+    }
+
+    fn blacklist_bad_nodes(&mut self) {
+        unsafe {
+            if !self::LOCAL_BLACKLISTS {
+                return;
+            }
+        }
+
+        let in_flight = self.requests_in_flight.clone();
+        in_flight
+            .iter()
+            .map(|(_request_id, node_id)| node_id)
+            .counts()
+            .iter()
+            .for_each(|(node_id_request, count)| {
+                let node_id1 = (*node_id_request).clone();
+                if count.clone() > 10 {
+                    // made 10 requests to this node
+                    // remove the node from requests_in_flight to reset the counter
+                    // and blacklist the node
+                    log::warn!("blacklisting {node_id1}.");
+                    let request_ids = self
+                        .requests_in_flight
+                        .clone()
+                        .iter()
+                        .filter(|(_request_id, node_id2)| node_id1 == **node_id2)
+                        .map(|(request_id, _node_id)| request_id)
+                        .copied()
+                        .collect_vec();
+                    for value in request_ids {
+                        self.requests_in_flight.remove_entry(&value);
+                    }
+                    self.core.remove_node(&node_id1);
+                    self.blacklisted_nodes.insert(node_id1);
+                }
+            });
+    }
+
+    fn randomly_whitelist(&mut self) {
+        unsafe {
+            if !self::LOCAL_BLACKLISTS {
+                return;
+            }
+        }
+
+        let mut rng = rand::thread_rng();
+        if !self.blacklisted_nodes.is_empty() && rng.gen_bool(0.05) {
+            let len = self.blacklisted_nodes.len();
+            if len > 0 {
+                let random_index = rng.gen_range(0..len);
+                let node_id_opt = self.blacklisted_nodes.iter().nth(random_index).clone();
+                if let Some(node_id) = node_id_opt {
+                    log::warn!("whitelisting {node_id}.");
+                    self.core.add_node(node_id.clone());
+                    self.blacklisted_nodes.remove(&(node_id.clone()));
+                }
+            }
+        }
+    }
+
+    fn log_requests_in_flight(&self) {
+        log::info!("counts:");
+        self.requests_in_flight
+            .iter()
+            .map(|tuple| tuple.1)
+            .counts()
+            .iter()
+            .for_each(|item| log::info!("    {} -> {}", item.0, item.1))
+    }
+
     fn message_closest(
-        &self,
+        &mut self,
         orig: NodeID,
         last_hop: NodeID,
         key: U256,
         msg: NetworkWrapper,
     ) -> Vec<InternOut> {
-        match self
+        let readflo_id_opt = self.get_readflo_id(msg.clone());
+
+        self.blacklist_bad_nodes();
+
+        let closest = self
             .closest_or_connected(key.clone(), Some(&last_hop))
             .first()
-        {
-            Some(&next_hop) => vec![
-                ModuleMessage::Closest(orig, key, msg.clone()).wrapper_network(next_hop),
-                DHTRouterOut::MessageRouting(orig, last_hop, next_hop, key, msg).into(),
-            ],
+            .copied();
+
+        self.randomly_whitelist();
+
+        match closest.clone() {
+            Some(next_hop) => {
+                if self.blacklisted_nodes.contains(&next_hop) {
+                    log::error!("IMPOSSIBLE?: sending a message to a blacklisted node.");
+                }
+
+                if let Some(readflo_id) = readflo_id_opt {
+                    log::info!("NEXT HOP: {}", next_hop);
+                    self.requests_in_flight
+                        .insert(readflo_id.clone(), next_hop.clone());
+                    self.log_requests_in_flight();
+                }
+                vec![
+                    ModuleMessage::Closest(orig, key, msg.clone()).wrapper_network(next_hop),
+                    DHTRouterOut::MessageRouting(orig, last_hop, next_hop, key, msg).into(),
+                ]
+            }
             None => {
+                if readflo_id_opt.is_some() {
+                    log::warn!("NO NEXT HOP!");
+                }
                 if key == self.core.root {
                     vec![DHTRouterOut::MessageDest(orig, last_hop, msg).into()]
                 } else {
diff --git a/flmodules/src/dht_storage/messages.rs b/flmodules/src/dht_storage/messages.rs
@@ -47,7 +47,8 @@ pub enum MessageClosest {
     // which have enough place left.
     StoreFlo(Flo),
     // Request a Flo. The FloID is in the DHTRouting::Request.
-    ReadFlo(RealmID),
+    // The U256 is a random id identifying this specific request
+    ReadFlo(RealmID, U256),
     // Request Cuckoos for the given ID. The FloID is in the DHTRouting::Request.
     GetCuckooIDs(RealmID),
     // Store the Cuckoo-ID in the relevant Flo. The DHTRouting::Request(key) is the
@@ -61,7 +62,8 @@ pub enum MessageClosest {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub enum MessageDest {
     // Returns the result of the requested Flo, including any available Cuckoo-IDs.
-    FloValue(FloCuckoo),
+    // The optional u256 is an optional identifier of the readflo request id
+    FloValue(FloCuckoo, Option<U256>),
     // Indicates this Flo is not in the closest node.
     UnknownFlo(GlobalID),
     // The Cuckoo-IDs stored next to the Flo composed of the GlobalID
@@ -154,6 +156,7 @@ pub struct Messages {
 }
 
 pub static mut EVIL_NO_FORWARD: bool = false;
+pub static mut LOCAL_BLACKLISTS: bool = false;
 
 impl Messages {
     /// Returns a new simulation_chat module.
@@ -213,7 +216,7 @@ impl Messages {
             DHTStorageIn::StoreFlo(flo) => self.store_flo(flo),
             DHTStorageIn::ReadFlo(id) => vec![match self.read_flo(&id) {
                 Some(df) => DHTStorageOut::FloValue(df.clone()).into(),
-                None => MessageClosest::ReadFlo(id.realm_id().clone())
+                None => MessageClosest::ReadFlo(id.realm_id().clone(), U256::rnd())
                     .to_intern_out(id.flo_id().clone().into())
                     // .inspect(|msg| log::info!("{} sends {msg:?}", self.our_id))
                     .expect("Creating ReadFlo message"),
@@ -262,7 +265,7 @@ impl Messages {
             MessageClosest::StoreFlo(flo) => {
                 return self.store_flo(flo);
             }
-            MessageClosest::ReadFlo(rid) => {
+            MessageClosest::ReadFlo(rid, request_id) => {
                 // log::info!(
                 //     "{} got request for {}/{} from {}",
                 //     self.our_id,
@@ -279,7 +282,7 @@ impl Messages {
                     // log::info!("sends flo {:?}", fc.0);
                     if unsafe { !EVIL_NO_FORWARD } {
                         increment_stat!(self, self.experiment_stats.flo_value_sent_total);
-                        return MessageDest::FloValue(fc)
+                        return MessageDest::FloValue(fc, Some(request_id))
                             .to_intern_out(origin)
                             // .inspect(|msg| log::info!("{} sends {msg:?}", self.our_id))
                             .map_or(vec![], |msg| vec![msg]);
@@ -312,7 +315,7 @@ impl Messages {
 
     fn msg_dest(&mut self, msg: MessageDest) -> Vec<InternOut> {
         match msg {
-            MessageDest::FloValue(fc) => {
+            MessageDest::FloValue(fc, _) => {
                 // log::info!("{} stores {:?}", self.our_id, flo.0);
                 self.store_flo(fc.0.clone());
                 self.realms
@@ -723,11 +726,11 @@ mod tests {
 
         let out = NetworkWrapper::wrap_yaml(
             MODULE_NAME,
-            &MessageDest::FloValue((fr.flo().clone(), vec![])),
+            &MessageDest::FloValue((fr.flo().clone(), vec![]), None),
         )
         .unwrap();
 
-        if let MessageDest::FloValue(flo) = out.unwrap_yaml(MODULE_NAME).unwrap() {
+        if let MessageDest::FloValue(flo, _) = out.unwrap_yaml(MODULE_NAME).unwrap() {
             let fr2 = TryInto::<FloRealm>::try_into(flo.0)?;
             assert_eq!(fr, fr2);
         } else {

Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,12 @@ pub enum SimulationSubcommand {`
`63`	`63`
`64`	`64`	`#[arg(long)]`
`65`	`65`	`experiment_id: u32,`
	`66`	`+`
	`67`	`+ #[arg(long, default_value = "false")]`
	`68`	`+ with_local_blacklists: bool,`
`66`	`69`	`},`
	`70`	`+`
	`71`	`+ JustFetchOnce {},`
`67`	`72`	`}`
`68`	`73`
`69`	`74`	`pub struct SimulationHandler {}`
`@@ -108,6 +113,7 @@ impl SimulationHandler {`
`108`	`113`	`timeout_ms,`
`109`	`114`	`enable_sync,`
`110`	`115`	`experiment_id,`
	`116`	`+ with_local_blacklists,`
`111`	`117`	`} => {`
`112`	`118`	`let evil_noforward = f.args.evil_noforward.clone();`
`113`	`119`	`SimulationDht::run_fetch_pages(`
`@@ -118,9 +124,11 @@ impl SimulationHandler {`
`118`	`124`	`timeout_ms,`
`119`	`125`	`experiment_id,`
`120`	`126`	`evil_noforward,`
	`127`	`+ with_local_blacklists,`
`121`	`128`	`)`
`122`	`129`	`.await`
`123`	`130`	`}`
	`131`	`+ SimulationSubcommand::JustFetchOnce {} => SimulationDht::just_fetch_once(f).await,`
`124`	`132`	`}`
`125`	`133`	`}`
`126`	`134`	`}`