From 258ca498caa465946cf51bddaff9a5a4bfbdea7e Mon Sep 17 00:00:00 2001 From: rustonbsd Date: Tue, 21 Apr 2026 23:42:06 +0200 Subject: [PATCH 1/4] fix: topic reference drop if topic.gossip_sender/receiver() is called instead of split (everything stops) + initial_interval added to bubble and message_overlap workers + reduced custom max_join_peers for merge strategies (now individually configurable) + added bash script to run many instances at once in tmux --- README.md | 4 + examples/full_config.rs | 4 + src/config.rs | 92 +++++++++++++++++++++-- src/gossip/merge/bubble.rs | 3 +- src/gossip/merge/message_overlap.rs | 4 +- src/gossip/topic/topic.rs | 112 +++++++++++++++++++++++++++- test_many_chats.sh | 20 +++++ 7 files changed, 226 insertions(+), 13 deletions(-) create mode 100755 test_many_chats.sh diff --git a/README.md b/README.md index 23ef733..38c7f67 100644 --- a/README.md +++ b/README.md @@ -238,16 +238,20 @@ Config::builder() .bubble_merge( BubbleMergeConfig::builder() .min_neighbors(4) + .initial_interval(Duration::from_secs(30)) .base_interval(Duration::from_secs(60)) .max_jitter(Duration::from_secs(120)) .fail_topic_creation_on_merge_startup_failure(true) + .max_join_peers(2) .build(), ) .message_overlap_merge( MessageOverlapMergeConfig::builder() + .initial_interval(Duration::from_secs(30)) .base_interval(Duration::from_secs(60)) .max_jitter(Duration::from_secs(120)) .fail_topic_creation_on_merge_startup_failure(true) + .max_join_peers(2) .build(), ) .build(), diff --git a/examples/full_config.rs b/examples/full_config.rs index e032550..1e24a28 100644 --- a/examples/full_config.rs +++ b/examples/full_config.rs @@ -49,16 +49,20 @@ fn config_builder() -> Config { .bubble_merge( BubbleMergeConfig::builder() .min_neighbors(4) + .initial_interval(Duration::from_secs(30)) .base_interval(Duration::from_secs(60)) .max_jitter(Duration::from_secs(120)) .fail_topic_creation_on_merge_startup_failure(true) + .max_join_peers(2) .build(), ) .message_overlap_merge( MessageOverlapMergeConfig::builder() + .initial_interval(Duration::from_secs(30)) .base_interval(Duration::from_secs(60)) .max_jitter(Duration::from_secs(120)) .fail_topic_creation_on_merge_startup_failure(true) + .max_join_peers(2) .build(), ) .build(), diff --git a/src/config.rs b/src/config.rs index 671fe91..270c69f 100644 --- a/src/config.rs +++ b/src/config.rs @@ -134,7 +134,7 @@ impl DhtConfigBuilder { } /// Timeout for DHT put operations. No-op if `timeout` is `Duration::ZERO`. - /// + /// /// If `put_timeout` is called only once with `Duration::ZERO`, default value prevails. /// If `put_timeout` is first called with a > `Duration::ZERO`, and then again with `Duration::ZERO`, the first set value is kept. /// @@ -147,7 +147,7 @@ impl DhtConfigBuilder { } /// Timeout for DHT get operations. No-op if `timeout` is `Duration::ZERO`. - /// + /// /// If `get_timeout` is called only once with `Duration::ZERO`, default value prevails. /// If `get_timeout` is first called with a > `Duration::ZERO`, and then again with `Duration::ZERO`, the first set value is kept. /// @@ -232,10 +232,12 @@ pub enum BubbleMergeConfig { #[derive(Debug, Clone)] pub struct BubbleMergeConfigInner { + initial_interval: Duration, base_interval: Duration, max_jitter: Duration, min_neighbors: usize, fail_topic_creation_on_merge_startup_failure: bool, + max_join_peers: usize, } #[derive(Debug, Clone)] @@ -252,10 +254,12 @@ impl Default for BubbleMergeConfig { impl Default for BubbleMergeConfigInner { fn default() -> Self { Self { + initial_interval: Duration::from_secs(30), base_interval: Duration::from_secs(60), max_jitter: Duration::from_secs(120), min_neighbors: 4, fail_topic_creation_on_merge_startup_failure: true, + max_join_peers: 2, } } } @@ -270,8 +274,15 @@ impl BubbleMergeConfig { } impl BubbleMergeConfigInner { + /// Initial delay before starting bubble merge attempts. + /// + /// Default: 30s. + pub fn initial_interval(&self) -> Duration { + self.initial_interval + } + /// Base interval for bubble merge attempts. - /// + /// /// `base_interval` > Duration::ZERO /// /// Default: 60s. @@ -301,9 +312,24 @@ impl BubbleMergeConfigInner { pub fn fail_topic_creation_on_merge_startup_failure(&self) -> bool { self.fail_topic_creation_on_merge_startup_failure } + + /// Max number of peers to join during a bubble merge attempt. + /// + /// Default: 2. + pub fn max_join_peers(&self) -> usize { + self.max_join_peers + } } impl BubbleMergeConfigBuilder { + /// Initial delay before starting bubble merge attempts. + /// + /// Default: 30s. + pub fn initial_interval(mut self, interval: Duration) -> Self { + self.config.initial_interval = interval; + self + } + /// Base interval for bubble merge attempts. No-op if `interval` is `Duration::ZERO`. /// /// If `base_interval` is called only once with `Duration::ZERO`, default value prevails. @@ -343,6 +369,19 @@ impl BubbleMergeConfigBuilder { self } + /// Max number of peers to join during a bubble merge attempt. No-op if `max_join_peers` is zero. + /// + /// If `max_join_peers` is called only once with zero, default value prevails. + /// If `max_join_peers` is first called with a > zero, and then again with zero, the first set value is kept. + /// + /// Default: 2. + pub fn max_join_peers(mut self, max_join_peers: usize) -> Self { + if max_join_peers > 0 { + self.config.max_join_peers = max_join_peers; + } + self + } + /// Build the `BubbleMergeConfig`. pub fn build(self) -> BubbleMergeConfig { BubbleMergeConfig::Enabled(self.config) @@ -358,9 +397,11 @@ pub enum MessageOverlapMergeConfig { #[derive(Debug, Clone)] pub struct MessageOverlapMergeConfigInner { + initial_interval: Duration, base_interval: Duration, max_jitter: Duration, fail_topic_creation_on_merge_startup_failure: bool, + max_join_peers: usize, } #[derive(Debug, Clone)] @@ -371,9 +412,11 @@ pub struct MessageOverlapMergeConfigBuilder { impl Default for MessageOverlapMergeConfigInner { fn default() -> Self { Self { + initial_interval: Duration::from_secs(30), base_interval: Duration::from_secs(60), max_jitter: Duration::from_secs(120), fail_topic_creation_on_merge_startup_failure: true, + max_join_peers: 2, } } } @@ -394,10 +437,17 @@ impl MessageOverlapMergeConfig { } impl MessageOverlapMergeConfigInner { + /// Initial delay before starting message overlap merge attempts. + /// + /// Default: 30s. + pub fn initial_interval(&self) -> Duration { + self.initial_interval + } + /// Base interval for message overlap merge attempts. /// /// `base_interval` > Duration::ZERO - /// + /// /// Default: 60s. pub fn base_interval(&self) -> Duration { self.base_interval @@ -418,9 +468,24 @@ impl MessageOverlapMergeConfigInner { pub fn fail_topic_creation_on_merge_startup_failure(&self) -> bool { self.fail_topic_creation_on_merge_startup_failure } + + /// Max number of peers to join during a message overlap merge attempt. + /// + /// Default: 2. + pub fn max_join_peers(&self) -> usize { + self.max_join_peers + } } impl MessageOverlapMergeConfigBuilder { + /// Initial delay before starting message overlap merge attempts. + /// + /// Default: 30s. + pub fn initial_interval(mut self, interval: Duration) -> Self { + self.config.initial_interval = interval; + self + } + /// Base interval for message overlap merge attempts. No-op if `interval` is `Duration::ZERO`. /// /// If `base_interval` is called only once with `Duration::ZERO`, default value prevails. @@ -452,6 +517,19 @@ impl MessageOverlapMergeConfigBuilder { self } + /// Max number of peers to join during a message overlap merge attempt. No-op if `max_join_peers` is zero. + /// + /// If `max_join_peers` is called only once with zero, default value prevails. + /// If `max_join_peers` is first called with a > zero, and then again with zero, the first set value is kept. + /// + /// Default: 2. + pub fn max_join_peers(mut self, max_join_peers: usize) -> Self { + if max_join_peers > 0 { + self.config.max_join_peers = max_join_peers; + } + self + } + /// Build the `MessageOverlapMergeConfig`. pub fn build(self) -> MessageOverlapMergeConfig { MessageOverlapMergeConfig::Enabled(self.config) @@ -515,7 +593,7 @@ impl PublisherConfigInner { /// Base interval for publisher attempts. /// /// `base_interval` > Duration::ZERO - /// + /// /// Default: 10s. pub fn base_interval(&self) -> Duration { self.base_interval @@ -860,8 +938,8 @@ impl Config { &self.bootstrap_config } - /// Max peers to join simultaneously. - /// + /// Max peers to join simultaneously. + /// /// Minimum is 1. /// /// Default: 4. diff --git a/src/gossip/merge/bubble.rs b/src/gossip/merge/bubble.rs index d014354..8e8eae1 100644 --- a/src/gossip/merge/bubble.rs +++ b/src/gossip/merge/bubble.rs @@ -42,12 +42,13 @@ impl BubbleMerge { gossip_receiver: GossipReceiver, cancel_token: tokio_util::sync::CancellationToken, max_join_peers: usize, + initial_interval: Duration, base_interval: Duration, max_jitter: Duration, min_neighbors: usize, ) -> Result { let base_interval = base_interval.max(Duration::from_secs(1)); - let mut ticker = tokio::time::interval(base_interval); + let mut ticker = tokio::time::interval_at(tokio::time::Instant::now() + initial_interval, base_interval); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); let api = Handle::spawn_with( diff --git a/src/gossip/merge/message_overlap.rs b/src/gossip/merge/message_overlap.rs index e9e3eb0..d6aca40 100644 --- a/src/gossip/merge/message_overlap.rs +++ b/src/gossip/merge/message_overlap.rs @@ -30,17 +30,19 @@ struct MessageOverlapMergeActor { impl MessageOverlapMerge { /// Create a new split-brain detector. + #[allow(clippy::too_many_arguments)] pub fn new( record_publisher: RecordPublisher, gossip_sender: GossipSender, gossip_receiver: GossipReceiver, cancel_token: tokio_util::sync::CancellationToken, max_join_peers: usize, + initial_interval: Duration, base_interval: Duration, max_jitter: Duration, ) -> Result { let base_interval = base_interval.max(Duration::from_secs(1)); - let mut ticker = tokio::time::interval(base_interval); + let mut ticker = tokio::time::interval_at(tokio::time::Instant::now() + initial_interval, base_interval); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); let api = Handle::spawn_with( diff --git a/src/gossip/topic/topic.rs b/src/gossip/topic/topic.rs index 4234bf6..9841523 100644 --- a/src/gossip/topic/topic.rs +++ b/src/gossip/topic/topic.rs @@ -133,15 +133,26 @@ impl Topic { /// Get the gossip sender for this topic. pub async fn gossip_sender(&self) -> Result { + let topic_ref = Arc::new(self.clone()); self.api - .call(act!(actor => actor.bootstrap.gossip_sender())) + .call(act!(actor => async move { + let mut sender = actor.bootstrap.gossip_sender().await?; + sender._topic_keep_alive = Some(topic_ref.clone()); + Ok(sender) + + })) .await } /// Get the gossip receiver for this topic. pub async fn gossip_receiver(&self) -> Result { + let topic_ref = Arc::new(self.clone()); self.api - .call(act!(actor => actor.bootstrap.gossip_receiver())) + .call(act!(actor => async move { + let mut receiver = actor.bootstrap.gossip_receiver().await?; + receiver._topic_keep_alive = Some(topic_ref.clone()); + Ok(receiver) + })) .await } @@ -290,7 +301,8 @@ impl TopicActor { self.bootstrap.gossip_sender().await?, self.bootstrap.gossip_receiver().await?, self.cancel_token.clone(), - self.record_publisher.config().max_join_peer_count(), + config.max_join_peers(), + config.initial_interval(), config.base_interval(), config.max_jitter(), config.min_neighbors(), @@ -331,7 +343,8 @@ impl TopicActor { self.bootstrap.gossip_sender().await?, self.bootstrap.gossip_receiver().await?, self.cancel_token.clone(), - self.record_publisher.config().max_join_peer_count(), + config.max_join_peers(), + config.initial_interval(), config.base_interval(), config.max_jitter(), ) @@ -473,4 +486,95 @@ mod tests { assert!(cancel_token.is_cancelled()); } + + #[tokio::test] + async fn test_topic_survives_topic_drop_split() { + let secret_key = iroh::SecretKey::generate(); + let signing_key = mainline::SigningKey::from_bytes(&secret_key.to_bytes()); + let endpoint = iroh::Endpoint::builder(iroh::endpoint::presets::N0) + .secret_key(secret_key.clone()) + .bind() + .await + .expect("failed to bind endpoint"); + let gossip = iroh_gossip::net::Gossip::builder().spawn(endpoint.clone()); + + let topic_id = crate::TopicId::new("my-iroh-gossip-topic-survives-drop-split".to_string()); + let initial_secret = b"my-initial-secret".to_vec(); + + let record_publisher = crate::RecordPublisher::new( + topic_id.clone(), + signing_key.clone(), + None, + initial_secret, + crate::config::Config::default(), + ); + + let topic = crate::Topic::new(record_publisher, gossip.clone(), true) + .await + .expect("failed to create Topic"); + + let cancel_token = topic.cancel_token(); + let (_sender, _receiver) = topic.split().await.expect("failed to split topic"); + + drop(topic); + + assert!( + tokio::time::timeout(std::time::Duration::from_secs(5), cancel_token.cancelled()) + .await + .is_err() + ); + + assert!(!cancel_token.is_cancelled()); + } + + #[tokio::test] + async fn test_topic_survives_topic_drop_manual_sender_receiver() { + let secret_key = iroh::SecretKey::generate(); + let signing_key = mainline::SigningKey::from_bytes(&secret_key.to_bytes()); + let endpoint = iroh::Endpoint::builder(iroh::endpoint::presets::N0) + .secret_key(secret_key.clone()) + .bind() + .await + .expect("failed to bind endpoint"); + let gossip = iroh_gossip::net::Gossip::builder().spawn(endpoint.clone()); + + let topic_id = crate::TopicId::new( + "my-iroh-gossip-topic-survives-drop-manual-sender-receiver".to_string(), + ); + let initial_secret = b"my-initial-secret".to_vec(); + + let record_publisher = crate::RecordPublisher::new( + topic_id.clone(), + signing_key.clone(), + None, + initial_secret, + crate::config::Config::default(), + ); + + let topic = crate::Topic::new(record_publisher, gossip.clone(), true) + .await + .expect("failed to create Topic"); + + let cancel_token = topic.cancel_token(); + let (_sender, _receiver) = ( + topic + .gossip_sender() + .await + .expect("failed to get gossip sender"), + topic + .gossip_receiver() + .await + .expect("failed to get gossip receiver"), + ); + + drop(topic); + + assert!( + tokio::time::timeout(std::time::Duration::from_secs(2), cancel_token.cancelled()) + .await + .is_err() + ); + + assert!(!cancel_token.is_cancelled()); + } } diff --git a/test_many_chats.sh b/test_many_chats.sh new file mode 100755 index 0000000..f7f0175 --- /dev/null +++ b/test_many_chats.sh @@ -0,0 +1,20 @@ +#!/bin/bash + +if [ $# -lt 1 ]; then + echo "Usage: $0 " + exit 1 +fi + +N=$1 +shift +CMD="cargo run --example chat" +SESSION="multi-$$" + +tmux new-session -d -s "$SESSION" "$CMD" + +for ((i = 2; i <= N; i++)); do + tmux split-window -t "$SESSION" "$CMD" + tmux select-layout -t "$SESSION" tiled +done + +tmux attach -t "$SESSION" \ No newline at end of file From 9a26dccad407028bd23b2b99d7df746eecd87a10 Mon Sep 17 00:00:00 2001 From: rustonbsd Date: Tue, 21 Apr 2026 23:59:45 +0200 Subject: [PATCH 2/4] chore: version bump v0.3.2 --- Cargo.lock | 2 +- Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3508a93..7396d7f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -761,7 +761,7 @@ dependencies = [ [[package]] name = "distributed-topic-tracker" -version = "0.3.1" +version = "0.3.2" dependencies = [ "actor-helper", "anyhow", diff --git a/Cargo.toml b/Cargo.toml index de77e01..0fa59be 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,7 +7,7 @@ repository = "https://github.com/rustonbsd/distributed-topic-tracker" readme = "README.md" keywords = ["networking"] categories = ["network-programming"] -version = "0.3.1" +version = "0.3.2" edition = "2024" [features] From aa2097071778fed6f2bc06055f4747e0ad9a9178 Mon Sep 17 00:00:00 2001 From: rustonbsd Date: Wed, 22 Apr 2026 00:04:35 +0200 Subject: [PATCH 3/4] fix: cleaned up new test (review impls) --- src/gossip/topic/topic.rs | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/src/gossip/topic/topic.rs b/src/gossip/topic/topic.rs index 9841523..b446c05 100644 --- a/src/gossip/topic/topic.rs +++ b/src/gossip/topic/topic.rs @@ -519,12 +519,23 @@ mod tests { drop(topic); assert!( - tokio::time::timeout(std::time::Duration::from_secs(5), cancel_token.cancelled()) + tokio::time::timeout(std::time::Duration::from_secs(2), cancel_token.cancelled()) .await .is_err() ); assert!(!cancel_token.is_cancelled()); + + drop(_sender); + drop(_receiver); + + assert!( + tokio::time::timeout(std::time::Duration::from_secs(2), cancel_token.cancelled()) + .await + .is_ok() + ); + + assert!(cancel_token.is_cancelled()); } #[tokio::test] @@ -576,5 +587,16 @@ mod tests { ); assert!(!cancel_token.is_cancelled()); + + drop(_sender); + drop(_receiver); + + assert!( + tokio::time::timeout(std::time::Duration::from_secs(2), cancel_token.cancelled()) + .await + .is_ok() + ); + + assert!(cancel_token.is_cancelled()); } } From ec2b0dbe636e78ea08cdd7b535deb10bd4c38811 Mon Sep 17 00:00:00 2001 From: rustonbsd Date: Wed, 22 Apr 2026 00:17:52 +0200 Subject: [PATCH 4/4] fix: double setting of topic ref in gossip_sender and gossip_receiver + some consistency cleanup --- src/crypto/record.rs | 26 +++++++++++++------------- src/dht.rs | 2 +- src/gossip/topic/bootstrap.rs | 11 ++++------- src/gossip/topic/publisher.rs | 4 ++-- src/gossip/topic/topic.rs | 33 ++++++++++++++------------------- 5 files changed, 34 insertions(+), 42 deletions(-) diff --git a/src/crypto/record.rs b/src/crypto/record.rs index d71581f..613c5a8 100644 --- a/src/crypto/record.rs +++ b/src/crypto/record.rs @@ -12,7 +12,7 @@ use serde::{Deserialize, Serialize}; use sha2::Digest; use tokio_util::sync::CancellationToken; -use crate::Config; +use crate::{Config, Dht, RotationHandle}; /// Topic identifier derived from a string via SHA512 hashing. /// @@ -131,14 +131,14 @@ impl RecordContent { /// Checks existing DHT record count before publishing to respect capacity limits. #[derive(Debug, Clone)] pub struct RecordPublisher { - dht: crate::dht::Dht, + dht: Dht, - config: crate::config::Config, + config: Config, topic_id: TopicId, pub_key: VerifyingKey, signing_key: SigningKey, - secret_rotation: Option, + secret_rotation: Option, initial_secret_hash: [u8; 32], } @@ -147,20 +147,20 @@ pub struct RecordPublisher { pub struct RecordPublisherBuilder { topic_id: TopicId, signing_key: SigningKey, - secret_rotation: Option, + secret_rotation: Option, initial_secret: Vec, - config: crate::config::Config, + config: Config, } impl RecordPublisherBuilder { /// Set a custom secret rotation strategy. - pub fn secret_rotation(mut self, secret_rotation: crate::crypto::keys::RotationHandle) -> Self { + pub fn secret_rotation(mut self, secret_rotation: RotationHandle) -> Self { self.secret_rotation = Some(secret_rotation); self } /// Set the configuration. - pub fn config(mut self, config: crate::config::Config) -> Self { + pub fn config(mut self, config: Config) -> Self { self.config = config; self } @@ -189,7 +189,7 @@ impl RecordPublisher { signing_key, secret_rotation: None, initial_secret: initial_secret.into(), - config: crate::config::Config::default(), + config: Config::default(), } } @@ -205,9 +205,9 @@ impl RecordPublisher { pub fn new( topic_id: impl Into, signing_key: SigningKey, - secret_rotation: Option, + secret_rotation: Option, initial_secret: impl Into>, - config: crate::config::Config, + config: Config, ) -> Self { let mut initial_secret_hash = sha2::Sha512::new(); initial_secret_hash.update(initial_secret.into()); @@ -216,7 +216,7 @@ impl RecordPublisher { .expect("hashing failed"); Self { - dht: crate::dht::Dht::new(config.dht_config()), + dht: Dht::new(config.dht_config()), config, topic_id: topic_id.into(), pub_key: signing_key.verifying_key(), @@ -261,7 +261,7 @@ impl RecordPublisher { } /// Get the secret rotation handle if set. - pub fn secret_rotation(&self) -> Option { + pub fn secret_rotation(&self) -> Option { self.secret_rotation.clone() } diff --git a/src/dht.rs b/src/dht.rs index 5b6c91c..a7c52f6 100644 --- a/src/dht.rs +++ b/src/dht.rs @@ -25,7 +25,7 @@ pub struct Dht { #[derive(Debug, Default)] struct DhtActor { dht: Option, - config: crate::config::DhtConfig, + config: DhtConfig, } impl Dht { diff --git a/src/gossip/topic/bootstrap.rs b/src/gossip/topic/bootstrap.rs index 62f7b82..6772ff8 100644 --- a/src/gossip/topic/bootstrap.rs +++ b/src/gossip/topic/bootstrap.rs @@ -9,10 +9,7 @@ use tokio::time::sleep; use tokio_util::sync::CancellationToken; use crate::{ - GossipSender, MAX_MESSAGE_HASHES, MAX_RECORD_PEERS, RecordPublisher, - config::BootstrapConfig, - crypto::Record, - gossip::{GossipRecordContent, receiver::GossipReceiver}, + GossipSender, MAX_MESSAGE_HASHES, MAX_RECORD_PEERS, RecordPublisher, TimeoutConfig, config::BootstrapConfig, crypto::Record, gossip::{GossipRecordContent, receiver::GossipReceiver} }; /// Manages the peer discovery and joining process. @@ -26,7 +23,7 @@ pub struct Bootstrap { #[derive(Debug)] struct BootstrapActor { - record_publisher: crate::crypto::RecordPublisher, + record_publisher: RecordPublisher, gossip_sender: GossipSender, gossip_receiver: GossipReceiver, cancel_token: tokio_util::sync::CancellationToken, @@ -36,10 +33,10 @@ struct BootstrapActor { impl Bootstrap { /// Create a new bootstrap process for a topic. pub async fn new( - record_publisher: crate::crypto::RecordPublisher, + record_publisher: RecordPublisher, gossip: iroh_gossip::net::Gossip, cancel_token: tokio_util::sync::CancellationToken, - timeout_config: crate::config::TimeoutConfig, + timeout_config: TimeoutConfig, bootstrap_config: BootstrapConfig, ) -> Result { let gossip_topic: iroh_gossip::api::GossipTopic = gossip diff --git a/src/gossip/topic/publisher.rs b/src/gossip/topic/publisher.rs index 621dc8e..1a2b7ca 100644 --- a/src/gossip/topic/publisher.rs +++ b/src/gossip/topic/publisher.rs @@ -3,7 +3,7 @@ use actor_helper::{Action, Handle, Receiver}; use std::time::Duration; -use crate::{GossipReceiver, MAX_MESSAGE_HASHES, MAX_RECORD_PEERS, RecordPublisher}; +use crate::{GossipReceiver, GossipRecordContent, MAX_MESSAGE_HASHES, MAX_RECORD_PEERS, RecordPublisher}; use anyhow::Result; /// Periodically publishes node state to DHT for peer discovery. @@ -122,7 +122,7 @@ impl PublisherActor { last_message_hashes.len() ); - let record_content = crate::gossip::GossipRecordContent { + let record_content = GossipRecordContent { active_peers: { let mut peers = [Default::default(); MAX_RECORD_PEERS]; peers[..active_peers.len()].copy_from_slice(&active_peers); diff --git a/src/gossip/topic/topic.rs b/src/gossip/topic/topic.rs index b446c05..28cc6f0 100644 --- a/src/gossip/topic/topic.rs +++ b/src/gossip/topic/topic.rs @@ -3,12 +3,10 @@ use std::sync::{Arc, Weak}; use crate::{ - BubbleMergeConfig, Config, GossipSender, MessageOverlapMergeConfig, - config::PublisherConfig, - gossip::{ + BubbleMergeConfig, Config, GossipReceiver, GossipSender, MessageOverlapMergeConfig, RecordPublisher, config::PublisherConfig, gossip::{ merge::{BubbleMerge, MessageOverlapMerge}, topic::{bootstrap::Bootstrap, publisher::Publisher}, - }, + } }; use actor_helper::{Handle, act, act_ok}; use anyhow::Result; @@ -30,7 +28,7 @@ struct TopicActor { publisher: Option, bubble_merge: Option, message_overlap_merge: Option, - record_publisher: crate::crypto::RecordPublisher, + record_publisher: RecordPublisher, cancel_token: CancellationToken, } @@ -49,7 +47,7 @@ impl Topic { /// * `gossip` - Gossip instance for topic subscription /// * `async_bootstrap` - If false, awaits until bootstrap completes pub async fn new( - record_publisher: crate::crypto::RecordPublisher, + record_publisher: RecordPublisher, gossip: iroh_gossip::net::Gossip, async_bootstrap: bool, ) -> Result { @@ -122,13 +120,8 @@ impl Topic { } /// Split into sender and receiver for message exchange. - pub async fn split(&self) -> Result<(GossipSender, crate::gossip::receiver::GossipReceiver)> { - let topic_ref = Arc::new(self.clone()); - let mut sender = self.gossip_sender().await?; - let mut receiver = self.gossip_receiver().await?; - sender._topic_keep_alive = Some(topic_ref.clone()); - receiver._topic_keep_alive = Some(topic_ref); - Ok((sender, receiver)) + pub async fn split(&self) -> Result<(GossipSender, GossipReceiver)> { + Ok((self.gossip_sender().await?, self.gossip_receiver().await?)) } /// Get the gossip sender for this topic. @@ -145,7 +138,7 @@ impl Topic { } /// Get the gossip receiver for this topic. - pub async fn gossip_receiver(&self) -> Result { + pub async fn gossip_receiver(&self) -> Result { let topic_ref = Arc::new(self.clone()); self.api .call(act!(actor => async move { @@ -157,7 +150,7 @@ impl Topic { } /// Get the record publisher for this topic. - pub async fn record_creator(&self) -> Result { + pub async fn record_creator(&self) -> Result { self.api .call(act_ok!(actor => async move { actor.record_publisher.clone() })) .await @@ -376,6 +369,8 @@ impl TopicActor { #[cfg(test)] mod tests { + use crate::RecordPublisher; + #[tokio::test] async fn test_receiver_returns_none_after_shutdown() { let secret_key = iroh::SecretKey::generate(); @@ -390,7 +385,7 @@ mod tests { let topic_id = crate::TopicId::new("shutdown-receiver-test".to_string()); let initial_secret = b"my-initial-secret".to_vec(); - let record_publisher = crate::RecordPublisher::new( + let record_publisher = RecordPublisher::new( topic_id.clone(), signing_key.clone(), None, @@ -458,7 +453,7 @@ mod tests { let topic_id = crate::TopicId::new("my-iroh-gossip-topic".to_string()); let initial_secret = b"my-initial-secret".to_vec(); - let record_publisher = crate::RecordPublisher::new( + let record_publisher = RecordPublisher::new( topic_id.clone(), signing_key.clone(), None, @@ -501,7 +496,7 @@ mod tests { let topic_id = crate::TopicId::new("my-iroh-gossip-topic-survives-drop-split".to_string()); let initial_secret = b"my-initial-secret".to_vec(); - let record_publisher = crate::RecordPublisher::new( + let record_publisher = RecordPublisher::new( topic_id.clone(), signing_key.clone(), None, @@ -554,7 +549,7 @@ mod tests { ); let initial_secret = b"my-initial-secret".to_vec(); - let record_publisher = crate::RecordPublisher::new( + let record_publisher = RecordPublisher::new( topic_id.clone(), signing_key.clone(), None,