Skip to content

fix: topic reference drop if topic.gossip_sender/receiver() is called… #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rustonbsd merged 4 commits into from
Apr 21, 2026
Merged

fix: topic reference drop if topic.gossip_sender/receiver() is called… #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
258ca49
fix: topic reference drop if topic.gossip_sender/receiver() is called…
rustonbsd Apr 21, 2026
9a26dcc
chore: version bump v0.3.2
rustonbsd Apr 21, 2026
aa20970
fix: cleaned up new test (review impls)
rustonbsd Apr 21, 2026
ec2b0db
fix: double setting of topic ref in gossip_sender and gossip_receiver…
rustonbsd Apr 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"

@coderabbitai coderabbitai Bot Apr 21, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Description: Check whether the changed merge constructors are exported through the crate's public module tree.
set -euo pipefail

fd -e rs '^(lib|mod)$|gossip|merge' src -x sh -c '
  echo "--- $1"
  rg -nP -C3 "pub\s+(mod|use)\s+.*(gossip|merge|BubbleMerge|MessageOverlapMerge)|\b(BubbleMerge|MessageOverlapMerge)\b" "$1" || true
' sh {}

Repository: rustonbsd/distributed-topic-tracker

Length of output: 61

🏁 Script executed:

# First, find all Rust files and understand directory structure
fd -e rs --type f src | head -20

Repository: rustonbsd/distributed-topic-tracker

Length of output: 61

🏁 Script executed:

# Search for BubbleMerge and MessageOverlapMerge definitions
rg "struct\s+(BubbleMerge|MessageOverlapMerge)" -t rust

Repository: rustonbsd/distributed-topic-tracker

Length of output: 542

🏁 Script executed:

# Check the public exports in lib.rs
cat src/lib.rs 2>/dev/null || echo "lib.rs not found"

Repository: rustonbsd/distributed-topic-tracker

Length of output: 1579

🏁 Script executed:

# Look for the merge module structure
fd -path "*/merge*" -t f src

Repository: rustonbsd/distributed-topic-tracker

Length of output: 247

🏁 Script executed:

# Find the new() constructor implementations for BubbleMerge
rg -A 5 "impl\s+BubbleMerge" src/gossip/merge/bubble.rs -t rust

Repository: rustonbsd/distributed-topic-tracker

Length of output: 771

🏁 Script executed:

# Find the new() constructor implementations for MessageOverlapMerge
rg -A 5 "impl\s+MessageOverlapMerge" src/gossip/merge/message_overlap.rs -t rust

Repository: rustonbsd/distributed-topic-tracker

Length of output: 848

🏁 Script executed:

# Look at the actual constructor signature for BubbleMerge
sed -n '/impl.*BubbleMerge/,/^}/p' src/gossip/merge/bubble.rs | head -30

Repository: rustonbsd/distributed-topic-tracker

Length of output: 1220

🏁 Script executed:

# Look at the actual constructor signature for MessageOverlapMerge
sed -n '/impl.*MessageOverlapMerge/,/^}/p' src/gossip/merge/message_overlap.rs | head -30

Repository: rustonbsd/distributed-topic-tracker

Length of output: 1214

Change version bump to 0.4.0 to reflect the breaking API change.

Adding initial_interval as a required positional argument to BubbleMerge::new() and MessageOverlapMerge::new() breaks backward compatibility. These constructors are publicly exported from lib.rs, so existing ^0.3 users will experience compilation failures. A patch bump (0.3.2) is incompatible with breaking changes; use a minor pre-1.0 bump such as 0.4.0.

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@Cargo.toml` at line 10, Update the crate version in Cargo.toml from "0.3.2"
to "0.4.0" to reflect the breaking API change introduced by adding the required
positional parameter `initial_interval` to the public constructors
`BubbleMerge::new()` and `MessageOverlapMerge::new()` (exported in lib.rs);
locate Cargo.toml and bump the version string to "0.4.0" so semantic versioning
correctly signals the incompatible change to users.

edition = "2024"

[features]
Expand Down
4 changes: 4 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand Down
4 changes: 4 additions & 0 deletions examples/full_config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand Down
92 changes: 85 additions & 7 deletions src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
///
Expand All @@ -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.
///
Expand Down Expand Up @@ -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)]
Expand All @@ -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,
}
}
}
Expand All @@ -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.
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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)
Expand All @@ -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)]
Expand All @@ -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,
}
}
}
Expand All @@ -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
Expand All @@ -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.
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -860,8 +938,8 @@ impl Config {
&self.bootstrap_config
}

/// Max peers to join simultaneously.
///
/// Max peers to join simultaneously.
///
/// Minimum is 1.
///
/// Default: 4.
Expand Down
26 changes: 13 additions & 13 deletions src/crypto/record.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
///
Expand Down Expand Up @@ -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<crate::crypto::keys::RotationHandle>,
secret_rotation: Option<RotationHandle>,
initial_secret_hash: [u8; 32],
}

Expand All @@ -147,20 +147,20 @@ pub struct RecordPublisher {
pub struct RecordPublisherBuilder {
topic_id: TopicId,
signing_key: SigningKey,
secret_rotation: Option<crate::crypto::keys::RotationHandle>,
secret_rotation: Option<RotationHandle>,
initial_secret: Vec<u8>,
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
}
Expand Down Expand Up @@ -189,7 +189,7 @@ impl RecordPublisher {
signing_key,
secret_rotation: None,
initial_secret: initial_secret.into(),
config: crate::config::Config::default(),
config: Config::default(),
}
}

Expand All @@ -205,9 +205,9 @@ impl RecordPublisher {
pub fn new(
topic_id: impl Into<TopicId>,
signing_key: SigningKey,
secret_rotation: Option<crate::crypto::keys::RotationHandle>,
secret_rotation: Option<RotationHandle>,
initial_secret: impl Into<Vec<u8>>,
config: crate::config::Config,
config: Config,
) -> Self {
let mut initial_secret_hash = sha2::Sha512::new();
initial_secret_hash.update(initial_secret.into());
Expand All @@ -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(),
Expand Down Expand Up @@ -261,7 +261,7 @@ impl RecordPublisher {
}

/// Get the secret rotation handle if set.
pub fn secret_rotation(&self) -> Option<crate::crypto::keys::RotationHandle> {
pub fn secret_rotation(&self) -> Option<RotationHandle> {
self.secret_rotation.clone()
}

Expand Down
2 changes: 1 addition & 1 deletion src/dht.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ pub struct Dht {
#[derive(Debug, Default)]
struct DhtActor {
dht: Option<mainline::async_dht::AsyncDht>,
config: crate::config::DhtConfig,
config: DhtConfig,
}

impl Dht {
Expand Down
3 changes: 2 additions & 1 deletion src/gossip/merge/bubble.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Self> {
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(
Expand Down
Loading