event_graph: remove read_confirms and unread_events

bin/darkirc/src/irc/server.rs

@@ -29,7 +29,7 @@ use log::{error, info};
 use darkfi::{
     event_graph::{
         model::{Event, EventId, ModelPtr},
-        protocol_event::{Seen, SeenPtr, UnreadEventsPtr},
+        protocol_event::{Seen, SeenPtr},
         view::ViewPtr,
     },
     net::P2pPtr,
@@ -52,7 +52,6 @@ pub struct IrcServer {
     p2p: P2pPtr,
     model: ModelPtr<PrivMsgEvent>,
     view: ViewPtr<PrivMsgEvent>,
-    unread_events: UnreadEventsPtr<PrivMsgEvent>,
     clients_subscriptions: SubscriberPtr<ClientSubMsg>,
     seen: SeenPtr<EventId>,
     missed_events: Arc<Mutex<Vec<Event<PrivMsgEvent>>>>,
@@ -66,7 +65,6 @@ impl IrcServer {
         p2p: P2pPtr,
         model: ModelPtr<PrivMsgEvent>,
         view: ViewPtr<PrivMsgEvent>,
-        unread_events: UnreadEventsPtr<PrivMsgEvent>,
         clients_subscriptions: SubscriberPtr<ClientSubMsg>,
     ) -> Result<Self> {
         let seen = Seen::new();
@@ -76,7 +74,6 @@ impl IrcServer {
             p2p,
             model,
             view,
-            unread_events,
             clients_subscriptions,
             seen,
             missed_events,
@@ -94,7 +91,6 @@ impl IrcServer {
                 self.p2p.clone(),
                 self.model.clone(),
                 self.seen.clone(),
-                self.unread_events.clone(),
                 msg_recv,
                 self.clients_subscriptions.clone(),
             ))
@@ -141,7 +137,6 @@ impl IrcServer {
         p2p: P2pPtr,
         model: ModelPtr<PrivMsgEvent>,
         seen: SeenPtr<EventId>,
-        unread_events: UnreadEventsPtr<PrivMsgEvent>,
         recv: smol::channel::Receiver<(NotifierMsg, u64)>,
         clients_subscriptions: SubscriberPtr<ClientSubMsg>,
     ) -> Result<()> {
@@ -169,7 +164,6 @@ impl IrcServer {
                         previous_event_hash: model.lock().await.get_head_hash(),
                         action: msg.clone(),
                         timestamp: Timestamp::current_time(),
-                        read_confirms: 0,
                     };
 
                     // Since this will be added to the View directly, other clients connected to irc
@@ -181,7 +175,6 @@ impl IrcServer {
                     if !seen.push(&event.hash()).await {
                         continue
                     }
-                    unread_events.lock().await.insert(&event);
 
                     p2p.broadcast(event).await?;
                 }

bin/darkirc/src/main.rs

@@ -33,7 +33,7 @@ use darkfi::{
     event_graph::{
         events_queue::EventsQueue,
         model::Model,
-        protocol_event::{ProtocolEvent, Seen, UnreadEvents},
+        protocol_event::{ProtocolEvent, Seen},
         view::View,
     },
     net,
@@ -101,8 +101,6 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'_>>) -> Result<(
     // Buffers
     let seen_event = Seen::new();
     let seen_inv = Seen::new();
-    let unread_events = UnreadEvents::new();
-    let unread_events_clone = unread_events.clone();
 
     // Check the version
     let mut net_settings = settings.net.clone();
@@ -119,10 +117,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'_>>) -> Result<(
             let seen_event = seen_event.clone();
             let seen_inv = seen_inv.clone();
             let model = model.clone();
-            let unread_events = unread_events.clone();
-            async move {
-                ProtocolEvent::init(channel, p2p, model, seen_event, seen_inv, unread_events).await
-            }
+            async move { ProtocolEvent::init(channel, p2p, model, seen_event, seen_inv).await }
         })
         .await;
 
@@ -154,7 +149,6 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'_>>) -> Result<(
         p2p.clone(),
         model_clone,
         view.clone(),
-        unread_events_clone,
         clients_subscriptions,
     )
     .await?;

bin/genev/genevd/src/main.rs

@@ -26,7 +26,7 @@ use darkfi::{
     event_graph::{
         events_queue::EventsQueue,
         model::{Event, EventId, Model},
-        protocol_event::{ProtocolEvent, Seen, SeenPtr, UnreadEvents},
+        protocol_event::{ProtocolEvent, Seen, SeenPtr},
         view::{View, ViewPtr},
     },
     net::{self, settings::SettingsOpt},
@@ -97,8 +97,6 @@ async fn realmain(args: Args, executor: Arc<smol::Executor<'_>>) -> Result<()> {
     // Buffers
     let seen_event = Seen::new();
     let seen_inv = Seen::new();
-    let unread_events = UnreadEvents::new();
-    let unread_events_clone = unread_events.clone();
 
     // Check the version
     let mut net_settings = args.net.clone();
@@ -115,10 +113,7 @@ async fn realmain(args: Args, executor: Arc<smol::Executor<'_>>) -> Result<()> {
             let seen_event = seen_event.clone();
             let seen_inv = seen_inv.clone();
             let model = model.clone();
-            let unread_events = unread_events.clone();
-            async move {
-                ProtocolEvent::init(channel, p2p, model, seen_event, seen_inv, unread_events).await
-            }
+            async move { ProtocolEvent::init(channel, p2p, model, seen_event, seen_inv).await }
         })
         .await;
 
@@ -142,7 +137,6 @@ async fn realmain(args: Args, executor: Arc<smol::Executor<'_>>) -> Result<()> {
     //
     let rpc_interface = Arc::new(JsonRpcInterface::new(
         "Alolymous".to_string(),
-        unread_events_clone,
         missed_events.clone(),
         model_clone,
         seen_ids.clone(),

bin/genev/genevd/src/rpc.rs

@@ -24,7 +24,7 @@ use serde_json::{json, Value};
 use darkfi::{
     event_graph::{
         model::{Event, EventId, ModelPtr},
-        protocol_event::{SeenPtr, UnreadEvents},
+        protocol_event::SeenPtr,
     },
     net,
     rpc::{
@@ -38,7 +38,6 @@ use crate::genevent::GenEvent;
 
 pub struct JsonRpcInterface {
     _nickname: String,
-    unread_events: Arc<Mutex<UnreadEvents<GenEvent>>>,
     missed_events: Arc<Mutex<Vec<Event<GenEvent>>>>,
     model: ModelPtr<GenEvent>,
     seen: SeenPtr<EventId>,
@@ -65,13 +64,12 @@ impl RequestHandler for JsonRpcInterface {
 impl JsonRpcInterface {
     pub fn new(
         _nickname: String,
-        unread_events: Arc<Mutex<UnreadEvents<GenEvent>>>,
         missed_events: Arc<Mutex<Vec<Event<GenEvent>>>>,
         model: ModelPtr<GenEvent>,
         seen: SeenPtr<EventId>,
         p2p: net::P2pPtr,
     ) -> Self {
-        Self { _nickname, unread_events, missed_events, model, seen, p2p }
+        Self { _nickname, missed_events, model, seen, p2p }
     }
 
     // RPCAPI:
@@ -106,7 +104,6 @@ impl JsonRpcInterface {
             previous_event_hash: self.model.lock().await.get_head_hash(),
             action: genevent,
             timestamp: Timestamp::current_time(),
-            read_confirms: 0,
         };
 
         if !self.seen.push(&event.hash()).await {
@@ -114,8 +111,6 @@ impl JsonRpcInterface {
             return JsonResponse::new(json, id).into()
         }
 
-        self.unread_events.lock().await.insert(&event);
-
         self.p2p.broadcast(event).await.unwrap();
 
         let json = json!(true);

bin/tau/taud/src/main.rs

@@ -42,7 +42,7 @@ use darkfi::{
     event_graph::{
         events_queue::EventsQueue,
         model::{Event, EventId, Model, ModelPtr},
-        protocol_event::{ProtocolEvent, Seen, SeenPtr, UnreadEvents},
+        protocol_event::{ProtocolEvent, Seen, SeenPtr},
         view::{View, ViewPtr},
         EventMsg,
     },
@@ -164,7 +164,6 @@ async fn start_sync_loop(
                         previous_event_hash: model.lock().await.get_head_hash(),
                         action: encrypted_task,
                         timestamp: Timestamp::current_time(),
-                        read_confirms: 0,
                     };
 
                     p2p.broadcast(event).await?;
@@ -320,7 +319,6 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'_>>) -> Result<(
     ////////////////////
     let seen_event = Seen::new();
     let seen_inv = Seen::new();
-    let unread_events = UnreadEvents::new();
 
     // let datastore_raft = datastore_path.join("tau.db");
 
@@ -341,10 +339,7 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'_>>) -> Result<(
             let seen_event = seen_event.clone();
             let seen_inv = seen_inv.clone();
             let model = model.clone();
-            let unread_events = unread_events.clone();
-            async move {
-                ProtocolEvent::init(channel, p2p, model, seen_event, seen_inv, unread_events).await
-            }
+            async move { ProtocolEvent::init(channel, p2p, model, seen_event, seen_inv).await }
         })
         .await;
 

src/event_graph/model.rs

@@ -37,7 +37,6 @@ pub struct Event<T: Send + Sync> {
     pub previous_event_hash: EventId,
     pub action: T,
     pub timestamp: Timestamp,
-    pub read_confirms: u8,
 }
 
 impl<T> Event<T>
@@ -46,10 +45,7 @@ where
 {
     pub fn hash(&self) -> EventId {
         let mut bytes = Vec::new();
-        let mut event_to_be_hashed = self.clone();
-        event_to_be_hashed.read_confirms = 0;
-        event_to_be_hashed.encode(&mut bytes).expect("serialize failed!");
-
+        self.encode(&mut bytes).expect("serialize failed!");
         let mut hasher = Ripemd256::new();
         hasher.update(bytes);
         let bytes = hasher.finalize().to_vec();
@@ -88,7 +84,6 @@ where
                 previous_event_hash: [0u8; 32],
                 action: T::new(),
                 timestamp: Timestamp(1674512021323),
-                read_confirms: 0,
             },
             children: Vec::new(),
         };
@@ -426,12 +421,7 @@ mod tests {
     }
 
     fn create_message(previous_event_hash: EventId, timestamp: u64) -> Event<PrivMsgEvent> {
-        Event {
-            previous_event_hash,
-            action: PrivMsgEvent::new(),
-            timestamp: Timestamp(timestamp),
-            read_confirms: 4,
-        }
+        Event { previous_event_hash, action: PrivMsgEvent::new(), timestamp: Timestamp(timestamp) }
     }
 
     /* THIS IS FAILING
@@ -622,14 +612,12 @@ mod tests {
 
         let timestamp = Timestamp::current_time().0 + 1;
         let event = create_message(root_id, timestamp);
-        let mut event2 = event.clone();
+        let event2 = event.clone();
 
         let event_hash = event.hash();
 
-        event2.read_confirms += 3;
         let event2_hash = event2.hash();
 
         assert_eq!(event2_hash, event_hash);
-        assert_ne!(event2.read_confirms, event.read_confirms);
     }
 }

src/event_graph/protocol_event.rs

@@ -37,7 +37,7 @@ use crate::{
 
 const UNREAD_EVENT_EXPIRE_TIME: u64 = 3600; // in seconds
 const SIZE_OF_SEEN_BUFFER: usize = 65536;
-const MAX_CONFIRM: u8 = 3;
+// const MAX_CONFIRM: u8 = 3;
 
 #[derive(Clone)]
 struct RingBuffer<T> {
@@ -121,7 +121,7 @@ where
         Arc::new(Mutex::new(Self { events: HashMap::new() }))
     }
 
-    fn contains(&self, key: &EventId) -> bool {
+    fn _contains(&self, key: &EventId) -> bool {
         self.events.contains_key(key)
     }
 
@@ -129,25 +129,6 @@ where
         self.events.get(key).cloned()
     }
 
-    // Increase the read_confirms for an event, if it has exceeded the MAX_CONFIRM
-    // then remove it from the hash_map and return Some(event), otherwise return None
-    fn inc_read_confirms(&mut self, key: &EventId) -> Option<Event<T>> {
-        let mut result = None;
-
-        if let Some(event) = self.events.get_mut(key) {
-            event.read_confirms += 1;
-            if event.read_confirms >= MAX_CONFIRM {
-                result = Some(event.clone())
-            }
-        }
-
-        if result.is_some() {
-            self.events.remove(key);
-        }
-
-        result
-    }
-
     pub fn insert(&mut self, event: &Event<T>) {
         // prune expired events
         let mut prune_ids = vec![];
@@ -178,7 +159,6 @@ where
     model: ModelPtr<T>,
     seen_event: SeenPtr<EventId>,
     seen_inv: SeenPtr<InvId>,
-    unread_events: UnreadEventsPtr<T>,
 }
 
 impl<T> ProtocolEvent<T>
@@ -191,7 +171,6 @@ where
         model: ModelPtr<T>,
         seen_event: SeenPtr<EventId>,
         seen_inv: SeenPtr<InvId>,
-        unread_events: UnreadEventsPtr<T>,
     ) -> net::ProtocolBasePtr {
         let message_subsytem = channel.get_message_subsystem();
         message_subsytem.add_dispatch::<Event<T>>().await;
@@ -224,7 +203,6 @@ where
             model,
             seen_event,
             seen_inv,
-            unread_events,
         })
     }
 
@@ -233,22 +211,13 @@ where
         // let exclude_list = vec![self.channel.address()];
         loop {
             let event = self.event_sub.receive().await?;
-            let mut event = (*event).to_owned();
-
-            // This could be better
+            let event = (*event).to_owned();
 
             if !self.seen_event.push(&event.hash()).await {
                 continue
             }
 
-            event.read_confirms += 1;
-
-            if event.read_confirms >= MAX_CONFIRM {
-                self.new_event(&event).await?;
-            } else {
-                self.unread_events.lock().await.insert(&event);
-            }
-
+            self.new_event(&event).await?;
             self.send_inv(&event).await?;
 
             // Broadcast the msg
@@ -269,17 +238,10 @@ where
                 continue
             }
 
-            {
-                let mut unread_events = self.unread_events.lock().await;
-
-                if !unread_events.contains(&inv_item.hash) &&
-                    self.model.lock().await.get_event(&inv_item.hash).is_none()
-                {
-                    self.send_getdata(vec![inv_item.hash]).await?;
-                } else if let Some(event) = unread_events.inc_read_confirms(&inv_item.hash) {
-                    self.new_event(&event).await?;
-                }
+            if self.model.lock().await.get_event(&inv_item.hash).is_none() {
+                self.send_getdata(vec![inv_item.hash]).await?;
             }
+
             // }
 
             // Broadcast the inv msg
@@ -293,12 +255,6 @@ where
             let events = (*getdata).to_owned().events;
 
             for event_id in events {
-                // let unread_event = self.unread_events.lock().await.get(&event_id);
-                // if let Some(event) = unread_event {
-                //     self.channel.send(event).await?;
-                //     continue
-                // }
-
                 let model_event = self.model.lock().await.get_event(&event_id);
                 if let Some(event) = model_event {
                     self.channel.send(event).await?;