pool: drop support for deprecated negentropy protocol

Closes #848

Pull-Request: #853
Signed-off-by: Yuki Kishimoto <[email protected]>

Author: yukibtc

CHANGELOG.md

@@ -27,6 +27,10 @@
 
 ## Unreleased
 
+### Breaking changes
+
+* pool: drop support for deprecated negentropy protocol ([Yuki Kishimoto] at https://github.com/rust-nostr/nostr/pull/853)
+
 ### Changed
 
 * Bump `lru` from 0.13.0 to 0.14.0 ([Yuki Kishimoto])

Cargo.lock

@@ -19,7 +19,6 @@ clap = "=4.4.18"
 js-sys = "0.3"
 lru = { version = "0.14", default-features = false }
 negentropy = { version = "0.5", default-features = false }
-negentropy-deprecated = { package = "negentropy", version = "0.3", default-features = false }
 nostr = { version = "0.41", path = "./crates/nostr", default-features = false }
 nostr-connect = { version = "0.41", path = "./crates/nostr-connect", default-features = false }
 nostr-database = { version = "0.41", path = "./crates/nostr-database", default-features = false }

Cargo.toml

@@ -22,7 +22,6 @@ async-wsocket = { workspace = true, features = ["socks"] }
 atomic-destructor.workspace = true
 lru.workspace = true
 negentropy = { workspace = true, features = ["std"] }
-negentropy-deprecated = { workspace = true, features = ["std"] }
 nostr = { workspace = true, features = ["std"] }
 nostr-database.workspace = true
 tokio = { workspace = true, features = ["macros", "sync"] }

crates/nostr-relay-pool/Cargo.toml

@@ -33,8 +33,6 @@ pub enum Error {
     Hex(hex::Error),
     /// Negentropy error
     Negentropy(negentropy::Error),
-    /// Negentropy error
-    NegentropyDeprecated(negentropy_deprecated::Error),
     /// Database error
     Database(DatabaseError),
     /// Generic timeout
@@ -133,7 +131,6 @@ impl fmt::Display for Error {
             Self::EventBuilder(e) => write!(f, "{e}"),
             Self::Hex(e) => write!(f, "{e}"),
             Self::Negentropy(e) => write!(f, "{e}"),
-            Self::NegentropyDeprecated(e) => write!(f, "{e}"),
             Self::Database(e) => write!(f, "{e}"),
             Self::Timeout => write!(f, "timeout"),
             Self::NotRepliedToPing => write!(f, "not replied to ping"),
@@ -234,12 +231,6 @@ impl From<negentropy::Error> for Error {
     }
 }
 
-impl From<negentropy_deprecated::Error> for Error {
-    fn from(e: negentropy_deprecated::Error) -> Self {
-        Self::NegentropyDeprecated(e)
-    }
-}
-
 impl From<DatabaseError> for Error {
     fn from(e: DatabaseError) -> Self {
         Self::Database(e)

crates/nostr-relay-pool/src/relay/error.rs

@@ -16,7 +16,6 @@ use async_wsocket::futures_util::{self, SinkExt, StreamExt};
 use async_wsocket::{ConnectionMode, Message};
 use atomic_destructor::AtomicDestroyer;
 use negentropy::{Id, Negentropy, NegentropyStorageVector};
-use negentropy_deprecated::{Bytes as BytesDeprecated, Negentropy as NegentropyDeprecated};
 use nostr::secp256k1::rand::{self, Rng};
 use nostr_database::prelude::*;
 use tokio::sync::mpsc::{self, Receiver, Sender};
@@ -1780,7 +1779,7 @@ impl InnerRelay {
 
     /// New negentropy protocol
     #[inline(never)]
-    pub(super) async fn sync_new(
+    pub(super) async fn sync(
         &self,
         filter: &Filter,
         items: Vec<(EventId, Timestamp)>,
@@ -1932,163 +1931,6 @@ impl InnerRelay {
 
         Ok(())
     }
-
-    /// Deprecated negentropy protocol
-    #[inline(never)]
-    pub(super) async fn sync_deprecated(
-        &self,
-        filter: &Filter,
-        items: Vec<(EventId, Timestamp)>,
-        opts: &SyncOptions,
-        output: &mut Reconciliation,
-    ) -> Result<(), Error> {
-        // Compose negentropy struct, add items and seal
-        let mut negentropy = NegentropyDeprecated::new(32, Some(NEGENTROPY_FRAME_SIZE_LIMIT))?;
-        for (id, timestamp) in items.into_iter() {
-            let id = BytesDeprecated::from_slice(id.as_bytes());
-            negentropy.add_item(timestamp.as_u64(), id)?;
-        }
-        negentropy.seal()?;
-
-        // Initiate message
-        let initial_message = negentropy.initiate()?;
-
-        // Subscribe to notifications
-        let mut notifications = self.internal_notification_sender.subscribe();
-        let mut temp_notifications = self.internal_notification_sender.subscribe();
-
-        // Send the initial negentropy message
-        let sub_id = SubscriptionId::generate();
-        let open_msg: ClientMessage = ClientMessage::NegOpen {
-            subscription_id: Cow::Borrowed(&sub_id),
-            filter: Cow::Borrowed(filter),
-            id_size: Some(32),
-            initial_message: Cow::Owned(hex::encode(initial_message)),
-        };
-        self.send_msg(open_msg)?;
-
-        // Check if negentropy is supported
-        check_negentropy_support(&sub_id, opts, &mut temp_notifications).await?;
-
-        let mut in_flight_up: HashSet<EventId> = HashSet::new();
-        let mut in_flight_down: bool = false;
-        let mut sync_done: bool = false;
-        let mut have_ids: Vec<EventId> = Vec::new();
-        let mut need_ids: Vec<EventId> = Vec::new();
-        let down_sub_id: SubscriptionId = SubscriptionId::generate();
-
-        // Start reconciliation
-        while let Ok(notification) = notifications.recv().await {
-            match notification {
-                RelayNotification::Message { message } => {
-                    match message {
-                        RelayMessage::NegMsg {
-                            subscription_id,
-                            message,
-                        } => {
-                            if subscription_id.as_ref() == &sub_id {
-                                let mut curr_have_ids: Vec<BytesDeprecated> = Vec::new();
-                                let mut curr_need_ids: Vec<BytesDeprecated> = Vec::new();
-
-                                // Parse message
-                                let query: BytesDeprecated =
-                                    BytesDeprecated::from_hex(message.as_ref())?;
-
-                                // Reconcile
-                                let msg: Option<BytesDeprecated> = negentropy.reconcile_with_ids(
-                                    &query,
-                                    &mut curr_have_ids,
-                                    &mut curr_need_ids,
-                                )?;
-
-                                // Handle the message
-                                self.handle_neg_msg(
-                                    &subscription_id,
-                                    msg.map(|m| m.to_bytes()),
-                                    curr_have_ids.into_iter().filter_map(neg_depr_to_event_id),
-                                    curr_need_ids.into_iter().filter_map(neg_depr_to_event_id),
-                                    opts,
-                                    output,
-                                    &mut have_ids,
-                                    &mut need_ids,
-                                    &mut sync_done,
-                                )?;
-                            }
-                        }
-                        RelayMessage::NegErr {
-                            subscription_id,
-                            message,
-                        } => {
-                            if subscription_id.as_ref() == &sub_id {
-                                return Err(Error::RelayMessage(message.into_owned()));
-                            }
-                        }
-                        RelayMessage::Ok {
-                            event_id,
-                            status,
-                            message,
-                        } => {
-                            self.handle_neg_ok(
-                                &mut in_flight_up,
-                                event_id,
-                                status,
-                                message,
-                                output,
-                            );
-                        }
-                        RelayMessage::Event {
-                            subscription_id,
-                            event,
-                        } => {
-                            if subscription_id.as_ref() == &down_sub_id {
-                                output.received.insert(event.id);
-                            }
-                        }
-                        RelayMessage::EndOfStoredEvents(id) => {
-                            if id.as_ref() == &down_sub_id {
-                                in_flight_down = false;
-                            }
-                        }
-                        RelayMessage::Closed {
-                            subscription_id, ..
-                        } => {
-                            if subscription_id.as_ref() == &down_sub_id {
-                                in_flight_down = false;
-                            }
-                        }
-                        _ => (),
-                    }
-
-                    // Send events
-                    self.upload_neg_events(&mut have_ids, &mut in_flight_up, opts)
-                        .await?;
-
-                    // Get events
-                    self.req_neg_events(&mut need_ids, &mut in_flight_down, &down_sub_id, opts)?;
-                }
-                RelayNotification::RelayStatus { status } => {
-                    if status.is_disconnected() {
-                        return Err(Error::NotConnected);
-                    }
-                }
-                RelayNotification::Shutdown => return Err(Error::ReceivedShutdown),
-                _ => (),
-            };
-
-            if sync_done
-                && have_ids.is_empty()
-                && need_ids.is_empty()
-                && in_flight_up.is_empty()
-                && !in_flight_down
-            {
-                break;
-            }
-        }
-
-        tracing::info!(url = %self.url, "Deprecated negentropy reconciliation terminated.");
-
-        Ok(())
-    }
 }
 
 /// Send WebSocket messages with timeout set to [WEBSOCKET_TX_TIMEOUT].
@@ -2114,11 +1956,6 @@ fn neg_id_to_event_id(id: Id) -> EventId {
     EventId::from_byte_array(id.to_bytes())
 }
 
-#[inline]
-fn neg_depr_to_event_id(id: BytesDeprecated) -> Option<EventId> {
-    EventId::from_slice(id.as_bytes()).ok()
-}
-
 fn prepare_negentropy_storage(
     items: Vec<(EventId, Timestamp)>,
 ) -> Result<NegentropyStorageVector, Error> {

crates/nostr-relay-pool/src/relay/inner.rs

@@ -713,22 +713,9 @@ impl Relay {
 
         let mut output: Reconciliation = Reconciliation::default();
 
-        match self
-            .inner
-            .sync_new(&filter, items.clone(), opts, &mut output)
-            .await
-        {
-            Ok(..) => {}
-            Err(e) => match e {
-                Error::NegentropyNotSupported
-                | Error::Negentropy(negentropy::Error::UnsupportedProtocolVersion) => {
-                    self.inner
-                        .sync_deprecated(&filter, items, opts, &mut output)
-                        .await?;
-                }
-                e => return Err(e),
-            },
-        }
+        self.inner
+            .sync(&filter, items.clone(), opts, &mut output)
+            .await?;
 
         Ok(output)
     }