Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6d0f8cd50d
|
||
|
|
b8e957fe33
|
||
|
|
b4cc826e74
|
||
|
|
63b59482fe
|
||
|
|
43b30952f8
|
||
|
|
599cfd14dc
|
No files matched your search
@@ -1,5 +1,65 @@
|
|||||||
# Changelog
|
# Changelog
|
||||||
|
|
||||||
|
## [2.0.1] - 2026-04-25
|
||||||
|
|
||||||
|
### Fixed
|
||||||
|
|
||||||
|
- **TUI quit no longer hangs on a misleading "updating DHT" screen.**
|
||||||
|
The 2.0.0 migration to `iroh-gossip-rendezvous` removed the
|
||||||
|
synchronous mainline-DHT publish that the old `Gossip::shutdown()`
|
||||||
|
performed (rendezvous entries age out instead), but the TUI flow
|
||||||
|
was unchanged: pressing `q` drew an "updating DHT…" overlay,
|
||||||
|
awaited `swarm.shutdown()`, and only then restored the terminal —
|
||||||
|
so the user sat on the stale message for as long as
|
||||||
|
`endpoint.close()` took to flush in-flight connections. Restore
|
||||||
|
the terminal the moment `q` is pressed, drop the overlay, and
|
||||||
|
cap the graceful-shutdown wait at 2 s so a slow endpoint close
|
||||||
|
can't stall the process.
|
||||||
|
|
||||||
|
## [2.0.0] - 2026-04-23
|
||||||
|
|
||||||
|
### Breaking
|
||||||
|
|
||||||
|
- **Partition-healing / DHT-rendezvous extracted to its own crate.**
|
||||||
|
The 700+ lines of gossip-partition-healing logic in `src/gossip.rs`
|
||||||
|
are now the `iroh-gossip-rendezvous` crate on crates.io (pinned
|
||||||
|
here at `^0.1`). Any downstream code that reached into
|
||||||
|
`bridge::gossip::*` for partition types needs to import from
|
||||||
|
`iroh_gossip_rendezvous::*` instead. The normative protocol spec
|
||||||
|
moved with the code — `dht-partition-healing.md` at the bridge
|
||||||
|
root is now a pointer to the new crate's `PROTOCOL.md`.
|
||||||
|
- **`iroh` and `iroh-gossip` bumped 0.97 → 0.98** (major; carries
|
||||||
|
iroh's own breaking changes through to any bridge consumers that
|
||||||
|
import from those crates transitively).
|
||||||
|
|
||||||
|
### Changed
|
||||||
|
|
||||||
|
- Direct deps `mainline`, `sha2`, `zeroize` removed — all now
|
||||||
|
reachable transitively via `iroh-gossip-rendezvous`. Closure is
|
||||||
|
smaller and the rendezvous concern lives in exactly one place.
|
||||||
|
- `donglora-client` version spec loosened `"1.0.0"` → `"1"`. Patch
|
||||||
|
and minor releases of the client propagate on the next
|
||||||
|
`cargo build` / `cargo install` without a bridge re-publish;
|
||||||
|
already-released `donglora-client 1.0.1` (CH340K USB bridge
|
||||||
|
support) is picked up automatically.
|
||||||
|
|
||||||
|
## [1.0.1] - 2026-04-22
|
||||||
|
|
||||||
|
### Security
|
||||||
|
|
||||||
|
- Bump both pinned `rand` versions past [RUSTSEC-2026-0097][adv] (`rand`
|
||||||
|
is unsound with a custom `log::Log` impl that calls `rand::rng()` at
|
||||||
|
trace/warn level and hits the 64 KB reseed path):
|
||||||
|
- `rand 0.9.2` → `0.9.4` (fix `>= 0.9.3`).
|
||||||
|
- `rand 0.8.5` → `0.8.6` (fix `>= 0.8.6`).
|
||||||
|
|
||||||
|
The advisory is only reachable under a very narrow set of conditions
|
||||||
|
(custom logger invoking `rand::rng()` during a reseed while emitting
|
||||||
|
trace-level logs, or warn-level with a failed `getrandom`). We are
|
||||||
|
not known to do that, but shipping the upgrade is the right call.
|
||||||
|
|
||||||
|
[adv]: https://rustsec.org/advisories/RUSTSEC-2026-0097
|
||||||
|
|
||||||
## [1.0.0] - 2026-04-22
|
## [1.0.0] - 2026-04-22
|
||||||
|
|
||||||
### Breaking
|
### Breaking
|
||||||
|
|||||||
Generated
+724
-337
File diff suppressed because it is too large.
Load diff
+5
-7
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "donglora-bridge"
|
name = "donglora-bridge"
|
||||||
version = "1.0.0"
|
version = "2.0.1"
|
||||||
edition = "2024"
|
edition = "2024"
|
||||||
authors = ["Stephen Waits <steve@waits.net>"]
|
authors = ["Stephen Waits <steve@waits.net>"]
|
||||||
license = "MIT"
|
license = "MIT"
|
||||||
@@ -14,10 +14,10 @@ rust-version = "1.94.1"
|
|||||||
exclude = [".jj/", ".github/"]
|
exclude = [".jj/", ".github/"]
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
donglora-client = "1.0.0"
|
donglora-client = "1"
|
||||||
iroh = "0.97"
|
iroh = "0.98"
|
||||||
iroh-gossip = "0.97"
|
iroh-gossip = "0.98"
|
||||||
mainline = { version = "6", features = ["async"] }
|
iroh-gossip-rendezvous = { version = "0.1", path = "../../iroh-gossip-rendezvous" }
|
||||||
tokio = { version = "1", features = ["full"] }
|
tokio = { version = "1", features = ["full"] }
|
||||||
clap = { version = "4", features = ["derive"] }
|
clap = { version = "4", features = ["derive"] }
|
||||||
serde = { version = "1", features = ["derive"] }
|
serde = { version = "1", features = ["derive"] }
|
||||||
@@ -34,10 +34,8 @@ bytes = "1"
|
|||||||
futures-lite = "2"
|
futures-lite = "2"
|
||||||
rand = "0.9"
|
rand = "0.9"
|
||||||
hex = "0.4.3"
|
hex = "0.4.3"
|
||||||
sha2 = "0.10"
|
|
||||||
tokio-util = { version = "0.7.18", features = ["rt"] }
|
tokio-util = { version = "0.7.18", features = ["rt"] }
|
||||||
jiff = "0.2.23"
|
jiff = "0.2.23"
|
||||||
zeroize = "1"
|
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
proptest = "1"
|
proptest = "1"
|
||||||
|
|||||||
@@ -0,0 +1,11 @@
|
|||||||
|
# Moved
|
||||||
|
|
||||||
|
The DHT partition-healing algorithm specification is now maintained as part
|
||||||
|
of the `iroh-gossip-rendezvous` crate:
|
||||||
|
|
||||||
|
**<https://github.com/swaits/iroh-gossip-rendezvous/blob/main/PROTOCOL.md>**
|
||||||
|
|
||||||
|
The bridge pins a specific `iroh-gossip-rendezvous` version; the spec that
|
||||||
|
matches whatever version you're running is in
|
||||||
|
`~/.cargo/registry/.../iroh-gossip-rendezvous-*/PROTOCOL.md` or, for path
|
||||||
|
dependencies, at `../iroh-gossip-rendezvous/PROTOCOL.md`.
|
||||||
+65
-677
@@ -1,140 +1,28 @@
|
|||||||
//! iroh-gossip swarm management and mainline DHT peer discovery.
|
//! Bridge-side glue over the `iroh-gossip-rendezvous` crate.
|
||||||
//!
|
//!
|
||||||
//! Derives cryptographic keys from a shared passphrase, manages the gossip
|
//! Starts the rendezvous, spawns a task that decodes incoming gossip events
|
||||||
//! subscription, and maintains a DHT heartbeat for zero-configuration peer
|
//! into [`GossipEvent`]s the router understands, and a broadcaster task that
|
||||||
//! bootstrapping.
|
//! encodes outgoing [`GossipFrame`]s onto the iroh-gossip sender.
|
||||||
|
|
||||||
use std::collections::HashSet;
|
use std::collections::HashSet;
|
||||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
use std::sync::Arc;
|
||||||
use std::sync::{Arc, Mutex};
|
|
||||||
use std::time::{Duration, Instant};
|
|
||||||
|
|
||||||
use anyhow::{Context, Result};
|
use anyhow::{Context, Result};
|
||||||
use bytes::Bytes;
|
use bytes::Bytes;
|
||||||
use futures_lite::StreamExt;
|
use iroh::{PublicKey, SecretKey};
|
||||||
use iroh::endpoint::Incoming;
|
|
||||||
use iroh::{Endpoint, PublicKey, SecretKey};
|
|
||||||
use iroh_gossip::TopicId;
|
|
||||||
use iroh_gossip::api::Event;
|
use iroh_gossip::api::Event;
|
||||||
use iroh_gossip::net::{GOSSIP_ALPN, Gossip as GossipActor};
|
use iroh_gossip_rendezvous::Rendezvous;
|
||||||
use mainline::async_dht::AsyncDht;
|
pub use iroh_gossip_rendezvous::{DhtStatus, RendezvousState};
|
||||||
use mainline::{Dht, MutableItem, SigningKey};
|
|
||||||
use sha2::{Digest, Sha256};
|
|
||||||
use tokio::sync::mpsc;
|
use tokio::sync::mpsc;
|
||||||
use tokio_util::sync::CancellationToken;
|
use tokio_util::sync::CancellationToken;
|
||||||
use tracing::{debug, error, info, warn};
|
use tracing::{debug, error, warn};
|
||||||
|
|
||||||
use crate::packet::GossipFrame;
|
use crate::packet::GossipFrame;
|
||||||
|
|
||||||
/// Maximum number of peers to store in a DHT record.
|
/// App salt for the donglora-bridge rendezvous surface. Versioned so a
|
||||||
/// 28 × 32 bytes = 896 bytes raw. Bencoded: 28 × ~36 = ~1008.
|
/// future breaking change gets a fresh DHT meeting place — v1 and v2
|
||||||
/// Conservative cap to stay under BEP44's 1000-byte value limit.
|
/// bridges do not see each other.
|
||||||
const MAX_DHT_PEERS: usize = 27;
|
pub const APP_SALT: &str = "donglora-bridge/v2";
|
||||||
|
|
||||||
/// DHT heartbeat interval (base, before jitter). Each bridge periodically
|
|
||||||
/// publishes `[self] + live_neighbors` as a blind overwrite. Any live bridge
|
|
||||||
/// publishing at this cadence guarantees that the DHT record is overwritten
|
|
||||||
/// with a fresh snapshot of a real, connected sub-swarm within this window —
|
|
||||||
/// so stale entries from crashed / restarted peers age out within one cycle.
|
|
||||||
const DHT_HEARTBEAT_SECS: u64 = 60;
|
|
||||||
|
|
||||||
/// Maximum random jitter added to heartbeat. Prevents synchronized publish
|
|
||||||
/// storms when multiple bridges start simultaneously.
|
|
||||||
const DHT_HEARTBEAT_JITTER_SECS: u64 = 15;
|
|
||||||
|
|
||||||
/// DHT read interval while we have no neighbors. Stays aggressive so two
|
|
||||||
/// freshly-started bridges find each other quickly.
|
|
||||||
const DHT_LONELY_READ_SECS: u64 = 3;
|
|
||||||
|
|
||||||
/// How long to try dialing DHT-discovered peers before we give up and publish
|
|
||||||
/// ourselves alone. If we connect to at least one peer in this window we
|
|
||||||
/// publish `[self, neighbors...]` right away. If we don't, we publish `[self]`
|
|
||||||
/// so newcomers find us — but we delay this so we don't overwrite a
|
|
||||||
/// potentially useful DHT record with just our own lonely entry.
|
|
||||||
const DHT_BOOTSTRAP_GRACE_SECS: u64 = 10;
|
|
||||||
|
|
||||||
// ── Key derivation ───────────────────────────────────────────────
|
|
||||||
|
|
||||||
/// Cryptographic keys derived from a shared passphrase.
|
|
||||||
pub struct PassphraseKeys {
|
|
||||||
pub topic_id: TopicId,
|
|
||||||
pub dht_signing_key: SigningKey,
|
|
||||||
pub dht_salt: [u8; 20],
|
|
||||||
}
|
|
||||||
|
|
||||||
impl PassphraseKeys {
|
|
||||||
/// Derive all keys from a passphrase.
|
|
||||||
///
|
|
||||||
/// Intermediate buffers containing the passphrase are zeroized after use to
|
|
||||||
/// minimize time sensitive material spends in memory.
|
|
||||||
#[must_use]
|
|
||||||
pub fn derive(passphrase: &str) -> Self {
|
|
||||||
// Application-specific prefix to avoid DHT collisions with other apps.
|
|
||||||
const APP_PREFIX: &[u8] = b"donglora-bridge:";
|
|
||||||
|
|
||||||
// Helper: build a zeroized buffer of [prefix | suffix | passphrase].
|
|
||||||
let keyed_buf = |suffix: &[u8]| -> Vec<u8> {
|
|
||||||
let mut buf = Vec::with_capacity(APP_PREFIX.len() + suffix.len() + passphrase.len());
|
|
||||||
buf.extend_from_slice(APP_PREFIX);
|
|
||||||
buf.extend_from_slice(suffix);
|
|
||||||
buf.extend_from_slice(passphrase.as_bytes());
|
|
||||||
buf
|
|
||||||
};
|
|
||||||
|
|
||||||
// Topic ID: blake3 hash of prefixed passphrase.
|
|
||||||
let mut buf = keyed_buf(b"");
|
|
||||||
let topic_bytes = *blake3::hash(&buf).as_bytes();
|
|
||||||
let topic_id = TopicId::from_bytes(topic_bytes);
|
|
||||||
zeroize::Zeroize::zeroize(&mut buf);
|
|
||||||
|
|
||||||
// DHT signing key: ed25519 from SHA-256 of prefixed passphrase + "sign:".
|
|
||||||
let mut buf = keyed_buf(b"sign:");
|
|
||||||
let sign_seed = Sha256::digest(&buf);
|
|
||||||
let seed_bytes: [u8; 32] = sign_seed.into();
|
|
||||||
let dht_signing_key = SigningKey::from_bytes(&seed_bytes);
|
|
||||||
zeroize::Zeroize::zeroize(&mut buf);
|
|
||||||
|
|
||||||
// DHT salt: first 20 bytes of SHA-256 of prefixed passphrase + "salt:".
|
|
||||||
let mut buf = keyed_buf(b"salt:");
|
|
||||||
let salt_hash = Sha256::digest(&buf);
|
|
||||||
let mut dht_salt = [0u8; 20];
|
|
||||||
dht_salt.copy_from_slice(&salt_hash[..20]);
|
|
||||||
zeroize::Zeroize::zeroize(&mut buf);
|
|
||||||
|
|
||||||
Self { topic_id, dht_signing_key, dht_salt }
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// ── Swarm state (for TUI) ───────────────────────────────────────
|
|
||||||
|
|
||||||
/// Observable state of the gossip swarm.
|
|
||||||
#[derive(Debug, Clone)]
|
|
||||||
pub struct SwarmState {
|
|
||||||
pub topic_hash: String,
|
|
||||||
pub neighbor_count: usize,
|
|
||||||
pub dht_status: DhtStatus,
|
|
||||||
pub last_dht_publish: Option<Instant>,
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Status of the mainline DHT connection.
|
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
||||||
pub enum DhtStatus {
|
|
||||||
Bootstrapping,
|
|
||||||
Ready,
|
|
||||||
PublishFailed,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl std::fmt::Display for DhtStatus {
|
|
||||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
||||||
match self {
|
|
||||||
Self::Bootstrapping => write!(f, "Bootstrapping"),
|
|
||||||
Self::Ready => write!(f, "Ready"),
|
|
||||||
Self::PublishFailed => write!(f, "Failed"),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// ── Events to router ────────────────────────────────────────────
|
|
||||||
|
|
||||||
/// Events from the gossip swarm to the router.
|
/// Events from the gossip swarm to the router.
|
||||||
pub enum GossipEvent {
|
pub enum GossipEvent {
|
||||||
@@ -144,212 +32,89 @@ pub enum GossipEvent {
|
|||||||
NeighborChanged(usize),
|
NeighborChanged(usize),
|
||||||
}
|
}
|
||||||
|
|
||||||
// ── Gossip handle ───────────────────────────────────────────────
|
/// Handle held by the rest of the bridge. Owns the `Rendezvous` that in turn
|
||||||
|
/// owns all DHT + gossip background work. Dropping this stops everything.
|
||||||
/// Handle to the gossip swarm. Manages iroh endpoint, gossip subscription, and DHT.
|
|
||||||
pub struct Gossip {
|
pub struct Gossip {
|
||||||
endpoint: Endpoint,
|
rendezvous: Rendezvous,
|
||||||
actor: GossipActor,
|
|
||||||
// std::sync::Mutex (not tokio::sync::Mutex) because locks are never held across
|
|
||||||
// .await points. This avoids async-aware mutex overhead for brief critical sections.
|
|
||||||
state: Arc<Mutex<SwarmState>>,
|
|
||||||
neighbors: Arc<Mutex<HashSet<PublicKey>>>,
|
|
||||||
dht_signing_key: SigningKey,
|
|
||||||
dht_salt: [u8; 20],
|
|
||||||
cancel: CancellationToken,
|
cancel: CancellationToken,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Gossip {
|
impl Gossip {
|
||||||
/// Create and start the gossip network.
|
/// Spin up the gossip swarm + DHT rendezvous.
|
||||||
///
|
|
||||||
/// Spawns background tasks for connection acceptance, gossip event forwarding,
|
|
||||||
/// gossip broadcasting, and DHT heartbeat.
|
|
||||||
///
|
///
|
||||||
/// Returns `(handle, event_rx, frame_tx)`:
|
/// Returns `(handle, event_rx, frame_tx)`:
|
||||||
/// - `event_rx`: channel of events from the swarm to the router
|
/// - `event_rx`: channel of events from the swarm to the router
|
||||||
/// - `frame_tx`: channel for the router to send frames to broadcast
|
/// - `frame_tx`: channel for the router to send frames to broadcast
|
||||||
pub async fn new(
|
pub async fn new(
|
||||||
secret_key: SecretKey,
|
secret_key: SecretKey,
|
||||||
keys: &PassphraseKeys,
|
passphrase: &str,
|
||||||
) -> Result<(Self, mpsc::Receiver<GossipEvent>, mpsc::Sender<GossipFrame>)> {
|
) -> Result<(Self, mpsc::Receiver<GossipEvent>, mpsc::Sender<GossipFrame>)> {
|
||||||
let cancel = CancellationToken::new();
|
let cancel = CancellationToken::new();
|
||||||
|
|
||||||
// Build iroh endpoint with n0 discovery.
|
let rendezvous = Rendezvous::builder()
|
||||||
let endpoint = Endpoint::builder(iroh::endpoint::presets::N0)
|
.passphrase(passphrase)
|
||||||
|
.app_salt(APP_SALT)
|
||||||
.secret_key(secret_key)
|
.secret_key(secret_key)
|
||||||
.alpns(vec![GOSSIP_ALPN.to_vec()])
|
.build()
|
||||||
.bind()
|
|
||||||
.await
|
.await
|
||||||
.context("binding iroh endpoint")?;
|
.context("starting iroh-gossip rendezvous")?;
|
||||||
|
|
||||||
let my_id = endpoint.id();
|
let my_id = rendezvous.node_id();
|
||||||
info!("iroh endpoint bound, id: {}", my_id.fmt_short());
|
tracing::info!("iroh endpoint bound, id: {}", my_id.fmt_short());
|
||||||
|
tracing::info!("topic: {}", hex::encode(&rendezvous.topic_id().as_bytes()[..8]));
|
||||||
|
|
||||||
// Build gossip actor.
|
|
||||||
let actor = GossipActor::builder().spawn(endpoint.clone());
|
|
||||||
|
|
||||||
// Bootstrap from DHT.
|
|
||||||
let bootstrap_peers = dht_get_peers(keys).await;
|
|
||||||
if bootstrap_peers.is_empty() {
|
|
||||||
info!("no peers found in DHT — we are the first node");
|
|
||||||
} else {
|
|
||||||
info!("found {} peer(s) in DHT", bootstrap_peers.len());
|
|
||||||
}
|
|
||||||
|
|
||||||
// Subscribe to gossip topic.
|
|
||||||
let topic = actor.subscribe(keys.topic_id, bootstrap_peers).await.context("subscribing to gossip topic")?;
|
|
||||||
|
|
||||||
let (sender, receiver) = topic.split();
|
|
||||||
|
|
||||||
// Initial swarm state.
|
|
||||||
let topic_hash = hex::encode(&keys.topic_id.as_bytes()[..8]);
|
|
||||||
let state = Arc::new(Mutex::new(SwarmState {
|
|
||||||
topic_hash,
|
|
||||||
neighbor_count: 0,
|
|
||||||
dht_status: DhtStatus::Bootstrapping,
|
|
||||||
last_dht_publish: None,
|
|
||||||
}));
|
|
||||||
let neighbor_count = Arc::new(AtomicUsize::new(0));
|
|
||||||
let neighbors: Arc<Mutex<HashSet<PublicKey>>> = Arc::new(Mutex::new(HashSet::new()));
|
|
||||||
|
|
||||||
// Channels.
|
|
||||||
let (event_tx, event_rx) = mpsc::channel::<GossipEvent>(256);
|
let (event_tx, event_rx) = mpsc::channel::<GossipEvent>(256);
|
||||||
let (frame_tx, frame_rx) = mpsc::channel::<GossipFrame>(256);
|
let (frame_tx, frame_rx) = mpsc::channel::<GossipFrame>(256);
|
||||||
|
|
||||||
let handle = Self {
|
// Receiver task: drain the rendezvous event stream, translate into
|
||||||
endpoint: endpoint.clone(),
|
// router-side `GossipEvent`s.
|
||||||
actor: actor.clone(),
|
spawn_receiver(rendezvous.subscribe(), event_tx, cancel.clone());
|
||||||
state: state.clone(),
|
|
||||||
neighbors: neighbors.clone(),
|
|
||||||
dht_signing_key: keys.dht_signing_key.clone(),
|
|
||||||
dht_salt: keys.dht_salt,
|
|
||||||
cancel: cancel.clone(),
|
|
||||||
};
|
|
||||||
|
|
||||||
// Spawn accept loop.
|
// Broadcaster task: pull encoded frames from the router, hand them
|
||||||
spawn_accept_loop(endpoint.clone(), actor.clone(), cancel.clone());
|
// to the iroh-gossip sender.
|
||||||
|
spawn_broadcaster(rendezvous.sender().clone(), frame_rx, cancel.clone());
|
||||||
|
|
||||||
// Spawn gossip receiver task.
|
Ok((Self { rendezvous, cancel }, event_rx, frame_tx))
|
||||||
spawn_receiver(receiver, event_tx, state.clone(), neighbor_count.clone(), neighbors.clone(), cancel.clone());
|
|
||||||
|
|
||||||
// Clone sender for DHT heartbeat to feed discovered peers via join_peers.
|
|
||||||
let dht_sender = sender.clone();
|
|
||||||
|
|
||||||
// Spawn gossip broadcaster task.
|
|
||||||
spawn_broadcaster(sender, frame_rx, cancel.clone());
|
|
||||||
|
|
||||||
// Spawn DHT heartbeat.
|
|
||||||
spawn_dht_heartbeat(
|
|
||||||
my_id,
|
|
||||||
keys.dht_signing_key.clone(),
|
|
||||||
keys.dht_salt,
|
|
||||||
dht_sender,
|
|
||||||
neighbors,
|
|
||||||
state,
|
|
||||||
neighbor_count,
|
|
||||||
cancel,
|
|
||||||
);
|
|
||||||
|
|
||||||
// Try to join (non-blocking — fire and forget with timeout).
|
|
||||||
// The receiver task handles events regardless.
|
|
||||||
|
|
||||||
Ok((handle, event_rx, frame_tx))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Snapshot of swarm state for TUI.
|
/// Snapshot of rendezvous state for the TUI.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn swarm_state(&self) -> SwarmState {
|
pub fn state(&self) -> RendezvousState {
|
||||||
self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner).clone()
|
self.rendezvous.state()
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Graceful shutdown. Publishes neighbors (without ourselves) to DHT
|
/// Graceful shutdown. Cancels our own glue tasks, then the rendezvous.
|
||||||
/// so other nodes can still bootstrap from live peers.
|
/// Idempotent.
|
||||||
pub async fn shutdown(&self) {
|
pub async fn shutdown(&self) {
|
||||||
// Publish neighbors-only to DHT before shutting down.
|
|
||||||
let neighbor_ids: Vec<[u8; 32]> = self
|
|
||||||
.neighbors
|
|
||||||
.lock()
|
|
||||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
|
||||||
.iter()
|
|
||||||
.map(|id| *id.as_bytes())
|
|
||||||
.collect();
|
|
||||||
|
|
||||||
if let Ok(dht) = Dht::client() {
|
|
||||||
let dht = dht.as_async();
|
|
||||||
let value = encode_peer_list(&neighbor_ids);
|
|
||||||
let seq = std::time::SystemTime::now()
|
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
|
||||||
.map(|d| d.as_secs().cast_signed())
|
|
||||||
.unwrap_or(0);
|
|
||||||
let item = MutableItem::new(self.dht_signing_key.clone(), &value, seq, Some(self.dht_salt.as_ref()));
|
|
||||||
match dht.put_mutable(item, None).await {
|
|
||||||
Ok(_) => info!("shutdown: published {} neighbor(s) to DHT (removed self)", neighbor_ids.len()),
|
|
||||||
Err(e) => debug!("shutdown: DHT publish failed: {e:#}"),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
self.cancel.cancel();
|
self.cancel.cancel();
|
||||||
let _ = self.actor.shutdown().await;
|
self.rendezvous.shutdown().await;
|
||||||
self.endpoint.close().await;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// ── Background tasks ────────────────────────────────────────────
|
// ── Internal tasks ──────────────────────────────────────────────────────
|
||||||
|
|
||||||
/// Accept incoming iroh connections and forward gossip ALPN to the gossip actor.
|
|
||||||
fn spawn_accept_loop(endpoint: Endpoint, gossip: GossipActor, cancel: CancellationToken) {
|
|
||||||
tokio::spawn(async move {
|
|
||||||
loop {
|
|
||||||
let incoming = tokio::select! {
|
|
||||||
incoming = endpoint.accept() => incoming,
|
|
||||||
() = cancel.cancelled() => break,
|
|
||||||
};
|
|
||||||
let Some(incoming) = incoming else { break };
|
|
||||||
let gossip = gossip.clone();
|
|
||||||
tokio::spawn(handle_incoming(incoming, gossip));
|
|
||||||
}
|
|
||||||
debug!("accept loop stopped");
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn handle_incoming(incoming: Incoming, gossip: GossipActor) {
|
|
||||||
let Ok(conn) = incoming.await else { return };
|
|
||||||
let alpn = conn.alpn();
|
|
||||||
if *alpn == *GOSSIP_ALPN {
|
|
||||||
if let Err(e) = gossip.handle_connection(conn).await {
|
|
||||||
debug!("gossip handle_connection error: {e:#}");
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
debug!("unexpected ALPN: {:?}", String::from_utf8_lossy(alpn));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Forward gossip events to the router via channel.
|
|
||||||
fn spawn_receiver(
|
fn spawn_receiver(
|
||||||
mut receiver: iroh_gossip::api::GossipReceiver,
|
mut rx: tokio::sync::broadcast::Receiver<Event>,
|
||||||
event_tx: mpsc::Sender<GossipEvent>,
|
event_tx: mpsc::Sender<GossipEvent>,
|
||||||
state: Arc<Mutex<SwarmState>>,
|
|
||||||
neighbor_count: Arc<AtomicUsize>,
|
|
||||||
neighbors: Arc<Mutex<HashSet<PublicKey>>>,
|
|
||||||
cancel: CancellationToken,
|
cancel: CancellationToken,
|
||||||
) {
|
) {
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
// Log a hard warning if the receiver stream ever ends — this is the
|
// Track our own neighbor set so we can send accurate counts to the
|
||||||
// "silent one-way gossip" failure mode. Track why it exited so the
|
// router (the rendezvous crate tracks the same set internally for
|
||||||
// log tells us whether it was a stream-end (None), an error
|
// its `state()`; we duplicate here to avoid a cross-crate handle).
|
||||||
// (Some(Err)), or a cancel.
|
let mut neighbors: HashSet<PublicKey> = HashSet::new();
|
||||||
let exit_reason: &'static str = loop {
|
let exit_reason: &'static str = loop {
|
||||||
let event = tokio::select! {
|
let event = tokio::select! {
|
||||||
event = receiver.next() => event,
|
ev = rx.recv() => ev,
|
||||||
() = cancel.cancelled() => break "cancelled",
|
() = cancel.cancelled() => break "cancelled",
|
||||||
};
|
};
|
||||||
let event = match event {
|
let event = match event {
|
||||||
None => break "stream ended (receiver closed)",
|
Ok(ev) => ev,
|
||||||
Some(Err(e)) => {
|
Err(tokio::sync::broadcast::error::RecvError::Closed) => break "broadcast closed",
|
||||||
error!("gossip receiver error: {e:#}");
|
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
|
||||||
break "receiver error";
|
warn!("gossip receiver lagged — dropped {n} events");
|
||||||
|
continue;
|
||||||
}
|
}
|
||||||
Some(Ok(ev)) => ev,
|
|
||||||
};
|
};
|
||||||
match event {
|
match event {
|
||||||
Event::Received(msg) => {
|
Event::Received(msg) => {
|
||||||
@@ -359,48 +124,31 @@ fn spawn_receiver(
|
|||||||
debug!("received malformed gossip frame ({} bytes)", msg.content.len());
|
debug!("received malformed gossip frame ({} bytes)", msg.content.len());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Neighbor count is tracked in both AtomicUsize (for lock-free reads in DHT
|
|
||||||
// heartbeat) and SwarmState (for TUI snapshots). Brief inconsistency between
|
|
||||||
// the two is acceptable — the TUI polls at 250ms intervals.
|
|
||||||
Event::NeighborUp(id) => {
|
Event::NeighborUp(id) => {
|
||||||
let count = neighbor_count.fetch_add(1, Ordering::Relaxed) + 1;
|
neighbors.insert(id);
|
||||||
info!("neighbor up: {} (total: {count})", id.fmt_short());
|
tracing::info!("neighbor up: {} (total: {})", id.fmt_short(), neighbors.len());
|
||||||
if let Ok(mut n) = neighbors.lock() {
|
let _ = event_tx.send(GossipEvent::NeighborChanged(neighbors.len())).await;
|
||||||
n.insert(id);
|
|
||||||
}
|
|
||||||
if let Ok(mut s) = state.lock() {
|
|
||||||
s.neighbor_count = count;
|
|
||||||
}
|
|
||||||
let _ = event_tx.send(GossipEvent::NeighborChanged(count)).await;
|
|
||||||
}
|
}
|
||||||
Event::NeighborDown(id) => {
|
Event::NeighborDown(id) => {
|
||||||
let count = neighbor_count.fetch_sub(1, Ordering::Relaxed).saturating_sub(1);
|
neighbors.remove(&id);
|
||||||
info!("neighbor down: {} (total: {count})", id.fmt_short());
|
tracing::info!("neighbor down: {} (total: {})", id.fmt_short(), neighbors.len());
|
||||||
if let Ok(mut n) = neighbors.lock() {
|
let _ = event_tx.send(GossipEvent::NeighborChanged(neighbors.len())).await;
|
||||||
n.remove(&id);
|
|
||||||
}
|
|
||||||
if let Ok(mut s) = state.lock() {
|
|
||||||
s.neighbor_count = count;
|
|
||||||
}
|
|
||||||
let _ = event_tx.send(GossipEvent::NeighborChanged(count)).await;
|
|
||||||
}
|
}
|
||||||
Event::Lagged => {
|
Event::Lagged => {
|
||||||
warn!("gossip receiver lagged — some messages were dropped");
|
warn!("gossip receiver lagged — some messages were dropped");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
// A shutdown cancel is expected; any other exit means we silently
|
|
||||||
// stopped receiving gossip. That's the "one-way bridge" bug —
|
|
||||||
// broadcast still works because spawn_broadcaster is independent.
|
|
||||||
if exit_reason == "cancelled" {
|
if exit_reason == "cancelled" {
|
||||||
debug!("gossip receiver stopped ({exit_reason})");
|
debug!("gossip receiver stopped ({exit_reason})");
|
||||||
} else {
|
} else {
|
||||||
error!("gossip receiver STOPPED ({exit_reason}) — this bridge will no longer receive from the swarm",);
|
error!(
|
||||||
|
"gossip receiver STOPPED ({exit_reason}) — this bridge will no longer receive from the swarm",
|
||||||
|
);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Read frames from the router and broadcast them to the swarm.
|
|
||||||
fn spawn_broadcaster(
|
fn spawn_broadcaster(
|
||||||
sender: iroh_gossip::api::GossipSender,
|
sender: iroh_gossip::api::GossipSender,
|
||||||
mut frame_rx: mpsc::Receiver<GossipFrame>,
|
mut frame_rx: mpsc::Receiver<GossipFrame>,
|
||||||
@@ -422,371 +170,11 @@ fn spawn_broadcaster(
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Periodically publish `[self] + live_neighbors` to the DHT and read it to
|
// Shared via Arc<Gossip> for the TUI.
|
||||||
/// discover new peers.
|
impl Gossip {
|
||||||
///
|
/// Wrap in `Arc` for sharing with the TUI task.
|
||||||
/// Strategy — deliberately simple:
|
|
||||||
/// 1. **Startup**: read DHT, try to join discovered peers. Do NOT publish
|
|
||||||
/// yet — we don't know if any of them are still alive, and publishing
|
|
||||||
/// `[self]` alone would overwrite a potentially useful record.
|
|
||||||
/// 2. **Bootstrap grace** (10 s): keep reading fast (every
|
|
||||||
/// `DHT_LONELY_READ_SECS`) looking for peers to connect to. If a
|
|
||||||
/// `NeighborUp` fires inside this window, publish immediately with the
|
|
||||||
/// live neighbor set. If nothing shows up, publish `[self]` so anyone
|
|
||||||
/// else joining finds us.
|
|
||||||
/// 3. **Steady state**: publish `[self, neighbors...]` every
|
|
||||||
/// `DHT_HEARTBEAT_SECS` (+ jitter), a blind overwrite. Any live bridge
|
|
||||||
/// doing this naturally ages out crashed/stale peers from the DHT
|
|
||||||
/// record within one cycle.
|
|
||||||
/// 4. **Re-read cadence**: fast (`DHT_LONELY_READ_SECS`) while
|
|
||||||
/// `neighbor_count == 0`, relaxed (= heartbeat interval) once
|
|
||||||
/// connected.
|
|
||||||
#[allow(clippy::too_many_arguments)]
|
|
||||||
fn spawn_dht_heartbeat(
|
|
||||||
my_id: PublicKey,
|
|
||||||
signing_key: SigningKey,
|
|
||||||
salt: [u8; 20],
|
|
||||||
gossip_sender: iroh_gossip::api::GossipSender,
|
|
||||||
neighbors: Arc<Mutex<HashSet<PublicKey>>>,
|
|
||||||
state: Arc<Mutex<SwarmState>>,
|
|
||||||
neighbor_count: Arc<AtomicUsize>,
|
|
||||||
cancel: CancellationToken,
|
|
||||||
) {
|
|
||||||
tokio::spawn(async move {
|
|
||||||
let dht = match Dht::client() {
|
|
||||||
Ok(d) => d.as_async(),
|
|
||||||
Err(e) => {
|
|
||||||
warn!("failed to create DHT client: {e:#}");
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
let pub_key_bytes = signing_key.verifying_key().to_bytes();
|
|
||||||
let mut last_publish: Option<Instant> = None;
|
|
||||||
let mut was_alone = true;
|
|
||||||
let started = Instant::now();
|
|
||||||
|
|
||||||
// Step 1: read the DHT to find peers to dial.
|
|
||||||
dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await;
|
|
||||||
|
|
||||||
loop {
|
|
||||||
let alone = neighbor_count.load(Ordering::Relaxed) == 0;
|
|
||||||
|
|
||||||
// Transition: alone → connected. Publish immediately with live
|
|
||||||
// neighbors so anyone who was about to dial us-as-lonely finds
|
|
||||||
// the full picture.
|
|
||||||
if was_alone && !alone {
|
|
||||||
info!("first neighbor joined — publishing to DHT");
|
|
||||||
dht_publish(&dht, &my_id, &signing_key, &salt, &neighbors, &state).await;
|
|
||||||
last_publish = Some(Instant::now());
|
|
||||||
}
|
|
||||||
was_alone = alone;
|
|
||||||
|
|
||||||
// Sleep: fast reads while lonely, relaxed once connected.
|
|
||||||
let sleep_dur = if alone {
|
|
||||||
Duration::from_secs(DHT_LONELY_READ_SECS)
|
|
||||||
} else {
|
|
||||||
let jitter = rand::random::<u64>() % DHT_HEARTBEAT_JITTER_SECS.max(1);
|
|
||||||
Duration::from_secs(DHT_HEARTBEAT_SECS + jitter)
|
|
||||||
};
|
|
||||||
|
|
||||||
tokio::select! {
|
|
||||||
() = tokio::time::sleep(sleep_dur) => {}
|
|
||||||
() = cancel.cancelled() => break,
|
|
||||||
}
|
|
||||||
|
|
||||||
// Always read — cheap, keeps fresh peers coming in.
|
|
||||||
dht_read_and_join(&dht, &my_id, &pub_key_bytes, &salt, &gossip_sender, &state).await;
|
|
||||||
|
|
||||||
// Steady-state: publish on every heartbeat tick once we've
|
|
||||||
// published at least once. Pre-first-publish, hold off until
|
|
||||||
// either a neighbor appears OR the bootstrap grace elapses —
|
|
||||||
// this avoids wiping a potentially-useful DHT record with
|
|
||||||
// just our lonely [self] entry on startup.
|
|
||||||
let should_publish = last_publish.map_or_else(
|
|
||||||
|| !alone || started.elapsed() >= Duration::from_secs(DHT_BOOTSTRAP_GRACE_SECS),
|
|
||||||
|t| t.elapsed() >= Duration::from_secs(DHT_HEARTBEAT_SECS),
|
|
||||||
);
|
|
||||||
if should_publish {
|
|
||||||
dht_publish(&dht, &my_id, &signing_key, &salt, &neighbors, &state).await;
|
|
||||||
last_publish = Some(Instant::now());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
debug!("DHT heartbeat stopped");
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Read the DHT and feed any discovered peers into the gossip swarm.
|
|
||||||
async fn dht_read_and_join(
|
|
||||||
dht: &AsyncDht,
|
|
||||||
my_id: &PublicKey,
|
|
||||||
pub_key_bytes: &[u8; 32],
|
|
||||||
salt: &[u8; 20],
|
|
||||||
gossip_sender: &iroh_gossip::api::GossipSender,
|
|
||||||
state: &Arc<Mutex<SwarmState>>,
|
|
||||||
) {
|
|
||||||
if let Some(record) = dht.get_mutable_most_recent(pub_key_bytes, Some(salt)).await {
|
|
||||||
let peers = decode_peer_list(record.value());
|
|
||||||
let new_peers: Vec<_> = peers.into_iter().filter(|id| *id != *my_id).collect();
|
|
||||||
if !new_peers.is_empty() {
|
|
||||||
info!("DHT discovered {} peer(s), feeding to gossip", new_peers.len());
|
|
||||||
if let Err(e) = gossip_sender.join_peers(new_peers).await {
|
|
||||||
debug!("join_peers error: {e:#}");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if let Ok(mut s) = state.lock() {
|
|
||||||
s.dht_status = DhtStatus::Ready;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Publish `[self] + live_neighbors` to the DHT as a blind overwrite.
|
|
||||||
///
|
|
||||||
/// No merge with existing DHT content. Stale peers age out naturally
|
|
||||||
/// because every live bridge publishes a snapshot of its actual live
|
|
||||||
/// connections within `DHT_HEARTBEAT_SECS`, and higher-seq records win.
|
|
||||||
async fn dht_publish(
|
|
||||||
dht: &AsyncDht,
|
|
||||||
my_id: &PublicKey,
|
|
||||||
signing_key: &SigningKey,
|
|
||||||
salt: &[u8; 20],
|
|
||||||
neighbors: &Arc<Mutex<HashSet<PublicKey>>>,
|
|
||||||
state: &Arc<Mutex<SwarmState>>,
|
|
||||||
) {
|
|
||||||
let mut peer_list: Vec<[u8; 32]> = Vec::with_capacity(MAX_DHT_PEERS);
|
|
||||||
peer_list.push(*my_id.as_bytes());
|
|
||||||
|
|
||||||
// Snapshot current live neighbors — those we have an active gossip
|
|
||||||
// connection to right now. No inherited junk from prior DHT content.
|
|
||||||
if let Ok(n) = neighbors.lock() {
|
|
||||||
for id in n.iter() {
|
|
||||||
if peer_list.len() >= MAX_DHT_PEERS {
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
peer_list.push(*id.as_bytes());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
let value = encode_peer_list(&peer_list);
|
|
||||||
let seq = std::time::SystemTime::now()
|
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
|
||||||
.map(|d| d.as_secs().cast_signed())
|
|
||||||
.unwrap_or(0);
|
|
||||||
|
|
||||||
let item = MutableItem::new(signing_key.clone(), &value, seq, Some(salt.as_ref()));
|
|
||||||
match dht.put_mutable(item, None).await {
|
|
||||||
Ok(_) => {
|
|
||||||
debug!("published to DHT (seq={seq}, peers={})", peer_list.len());
|
|
||||||
if let Ok(mut s) = state.lock() {
|
|
||||||
s.dht_status = DhtStatus::Ready;
|
|
||||||
s.last_dht_publish = Some(Instant::now());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Err(e) => {
|
|
||||||
warn!("DHT publish failed: {e:#}");
|
|
||||||
if let Ok(mut s) = state.lock() {
|
|
||||||
s.dht_status = DhtStatus::PublishFailed;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// ── DHT helpers ─────────────────────────────────────────────────
|
|
||||||
|
|
||||||
/// Query the DHT for existing peers (blocking bootstrap, called once on startup).
|
|
||||||
async fn dht_get_peers(keys: &PassphraseKeys) -> Vec<PublicKey> {
|
|
||||||
let dht = match Dht::client() {
|
|
||||||
Ok(d) => d.as_async(),
|
|
||||||
Err(e) => {
|
|
||||||
warn!("failed to create DHT client for bootstrap: {e:#}");
|
|
||||||
return vec![];
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
let pub_key_bytes = keys.dht_signing_key.verifying_key().to_bytes();
|
|
||||||
dht.get_mutable_most_recent(&pub_key_bytes, Some(&keys.dht_salt)).await.map_or_else(
|
|
||||||
|| {
|
|
||||||
debug!("DHT bootstrap: no existing record");
|
|
||||||
vec![]
|
|
||||||
},
|
|
||||||
|record| {
|
|
||||||
let peers = decode_peer_list(record.value());
|
|
||||||
debug!("DHT bootstrap: found {} peer(s)", peers.len());
|
|
||||||
peers
|
|
||||||
},
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Encode a list of 32-byte peer IDs into bytes for DHT storage.
|
|
||||||
/// Format: `[count: u16 LE] [id_0: 32B] [id_1: 32B] ...`
|
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn encode_peer_list(peers: &[[u8; 32]]) -> Vec<u8> {
|
pub fn into_arc(self) -> Arc<Self> {
|
||||||
#[allow(clippy::cast_possible_truncation)] // MAX_DHT_PEERS (27) fits in u16
|
Arc::new(self)
|
||||||
let count = peers.len().min(MAX_DHT_PEERS) as u16;
|
|
||||||
let mut buf = Vec::with_capacity(2 + (count as usize) * 32);
|
|
||||||
buf.extend_from_slice(&count.to_le_bytes());
|
|
||||||
for id in peers.iter().take(MAX_DHT_PEERS) {
|
|
||||||
buf.extend_from_slice(id);
|
|
||||||
}
|
|
||||||
buf
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Decode a peer list from DHT record value.
|
|
||||||
#[must_use]
|
|
||||||
pub fn decode_peer_list(data: &[u8]) -> Vec<PublicKey> {
|
|
||||||
if data.len() < 2 {
|
|
||||||
return vec![];
|
|
||||||
}
|
|
||||||
let count = u16::from_le_bytes([data[0], data[1]]) as usize;
|
|
||||||
let mut peers = Vec::with_capacity(count);
|
|
||||||
let mut offset = 2;
|
|
||||||
for _ in 0..count {
|
|
||||||
if offset + 32 > data.len() {
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
let bytes: [u8; 32] = match data[offset..offset + 32].try_into() {
|
|
||||||
Ok(b) => b,
|
|
||||||
Err(_) => break,
|
|
||||||
};
|
|
||||||
if let Ok(pk) = PublicKey::from_bytes(&bytes) {
|
|
||||||
peers.push(pk);
|
|
||||||
}
|
|
||||||
offset += 32;
|
|
||||||
}
|
|
||||||
peers
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
|
||||||
mod tests {
|
|
||||||
use super::*;
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn passphrase_keys_deterministic() {
|
|
||||||
let k1 = PassphraseKeys::derive("test-passphrase");
|
|
||||||
let k2 = PassphraseKeys::derive("test-passphrase");
|
|
||||||
assert_eq!(k1.topic_id, k2.topic_id);
|
|
||||||
assert_eq!(k1.dht_signing_key.to_bytes(), k2.dht_signing_key.to_bytes());
|
|
||||||
assert_eq!(k1.dht_salt, k2.dht_salt);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn passphrase_keys_different() {
|
|
||||||
let k1 = PassphraseKeys::derive("alpha");
|
|
||||||
let k2 = PassphraseKeys::derive("beta");
|
|
||||||
assert_ne!(k1.topic_id, k2.topic_id);
|
|
||||||
assert_ne!(k1.dht_signing_key.to_bytes(), k2.dht_signing_key.to_bytes());
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn peer_list_roundtrip() {
|
|
||||||
// Generate valid ed25519 public keys.
|
|
||||||
let keys: Vec<_> = (0..3)
|
|
||||||
.map(|_| {
|
|
||||||
let sk = iroh::SecretKey::generate(&mut rand::rng());
|
|
||||||
*sk.public().as_bytes()
|
|
||||||
})
|
|
||||||
.collect();
|
|
||||||
let encoded = encode_peer_list(&keys);
|
|
||||||
let decoded = decode_peer_list(&encoded);
|
|
||||||
assert_eq!(decoded.len(), 3);
|
|
||||||
for (i, key) in keys.iter().enumerate() {
|
|
||||||
assert_eq!(*decoded[i].as_bytes(), *key);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn peer_list_empty() {
|
|
||||||
let encoded = encode_peer_list(&[]);
|
|
||||||
let decoded = decode_peer_list(&encoded);
|
|
||||||
assert!(decoded.is_empty());
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn peer_list_truncated_data() {
|
|
||||||
// Only 1 byte — too short for count.
|
|
||||||
assert!(decode_peer_list(&[0]).is_empty());
|
|
||||||
// Count says 2 but only 1 peer present.
|
|
||||||
let mut data = vec![2, 0]; // count = 2
|
|
||||||
data.extend_from_slice(&[0xAA; 32]); // only 1 peer
|
|
||||||
let decoded = decode_peer_list(&data);
|
|
||||||
assert_eq!(decoded.len(), 1);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn dht_status_display() {
|
|
||||||
assert_eq!(format!("{}", DhtStatus::Bootstrapping), "Bootstrapping");
|
|
||||||
assert_eq!(format!("{}", DhtStatus::Ready), "Ready");
|
|
||||||
assert_eq!(format!("{}", DhtStatus::PublishFailed), "Failed");
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn decode_peer_list_exact_boundary() {
|
|
||||||
// Build data with count=2 and exactly 2 + 32 + 32 = 66 bytes.
|
|
||||||
// Both peers should decode successfully.
|
|
||||||
let sk0 = iroh::SecretKey::generate(&mut rand::rng());
|
|
||||||
let sk1 = iroh::SecretKey::generate(&mut rand::rng());
|
|
||||||
let id0 = *sk0.public().as_bytes();
|
|
||||||
let id1 = *sk1.public().as_bytes();
|
|
||||||
|
|
||||||
let mut data = vec![2, 0]; // count = 2 (u16 LE)
|
|
||||||
data.extend_from_slice(&id0);
|
|
||||||
data.extend_from_slice(&id1);
|
|
||||||
assert_eq!(data.len(), 66);
|
|
||||||
|
|
||||||
let decoded = decode_peer_list(&data);
|
|
||||||
assert_eq!(decoded.len(), 2);
|
|
||||||
assert_eq!(*decoded[0].as_bytes(), id0);
|
|
||||||
assert_eq!(*decoded[1].as_bytes(), id1);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn decode_peer_list_one_byte_short() {
|
|
||||||
// Build data with count=2 but only 2 + 32 + 31 = 65 bytes.
|
|
||||||
// First peer decodes, second is truncated and skipped.
|
|
||||||
let sk0 = iroh::SecretKey::generate(&mut rand::rng());
|
|
||||||
let sk1 = iroh::SecretKey::generate(&mut rand::rng());
|
|
||||||
let id0 = *sk0.public().as_bytes();
|
|
||||||
let id1 = *sk1.public().as_bytes();
|
|
||||||
|
|
||||||
let mut data = vec![2, 0]; // count = 2 (u16 LE)
|
|
||||||
data.extend_from_slice(&id0);
|
|
||||||
data.extend_from_slice(&id1[..31]); // only 31 bytes of second peer
|
|
||||||
assert_eq!(data.len(), 65);
|
|
||||||
|
|
||||||
let decoded = decode_peer_list(&data);
|
|
||||||
assert_eq!(decoded.len(), 1);
|
|
||||||
assert_eq!(*decoded[0].as_bytes(), id0);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[test]
|
|
||||||
fn decode_peer_list_count_header_only() {
|
|
||||||
// Data with exactly 2 bytes (count=0). Must not be rejected by the length check.
|
|
||||||
// Kills mutant: `data.len() < 2` → `data.len() <= 2`.
|
|
||||||
let data = [0u8, 0]; // count = 0
|
|
||||||
let decoded = decode_peer_list(&data);
|
|
||||||
assert!(decoded.is_empty());
|
|
||||||
|
|
||||||
// Data with 2 bytes but count=1 and no peer data — decodes 0 (truncated).
|
|
||||||
let data = [1u8, 0]; // count = 1 but no peer bytes
|
|
||||||
let decoded = decode_peer_list(&data);
|
|
||||||
assert!(decoded.is_empty());
|
|
||||||
}
|
|
||||||
|
|
||||||
mod proptests {
|
|
||||||
use super::*;
|
|
||||||
use proptest::prelude::*;
|
|
||||||
|
|
||||||
proptest! {
|
|
||||||
#[test]
|
|
||||||
fn peer_list_roundtrip_prop(count in 0usize..=30) {
|
|
||||||
let keys: Vec<[u8; 32]> = (0..count)
|
|
||||||
.map(|_| *iroh::SecretKey::generate(&mut rand::rng()).public().as_bytes())
|
|
||||||
.collect();
|
|
||||||
let encoded = encode_peer_list(&keys);
|
|
||||||
let decoded = decode_peer_list(&encoded);
|
|
||||||
let expected = count.min(MAX_DHT_PEERS);
|
|
||||||
prop_assert_eq!(decoded.len(), expected);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
+14
-13
@@ -1,7 +1,7 @@
|
|||||||
use donglora_bridge::{config, gossip, radio, rate_limit, router, setup, tui};
|
use donglora_bridge::{config, gossip, radio, rate_limit, router, setup, tui};
|
||||||
|
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::time::Instant;
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
use anyhow::{Context, Result};
|
use anyhow::{Context, Result};
|
||||||
use clap::{Parser, Subcommand};
|
use clap::{Parser, Subcommand};
|
||||||
@@ -10,7 +10,6 @@ use tokio::sync::mpsc;
|
|||||||
use tokio_util::sync::CancellationToken;
|
use tokio_util::sync::CancellationToken;
|
||||||
use tracing::{info, warn};
|
use tracing::{info, warn};
|
||||||
|
|
||||||
use crate::gossip::PassphraseKeys;
|
|
||||||
use crate::rate_limit::RateLimiter;
|
use crate::rate_limit::RateLimiter;
|
||||||
use crate::router::Stats;
|
use crate::router::Stats;
|
||||||
|
|
||||||
@@ -66,7 +65,7 @@ async fn main() -> Result<()> {
|
|||||||
setup_logging(cli.log_only, &cfg);
|
setup_logging(cli.log_only, &cfg);
|
||||||
|
|
||||||
// Ephemeral identity — fresh keypair each launch.
|
// Ephemeral identity — fresh keypair each launch.
|
||||||
let secret_key = SecretKey::generate(&mut rand::rng());
|
let secret_key = SecretKey::generate();
|
||||||
let node_id = secret_key.public();
|
let node_id = secret_key.public();
|
||||||
info!("donglora-bridge starting, ephemeral id: {}", node_id.fmt_short());
|
info!("donglora-bridge starting, ephemeral id: {}", node_id.fmt_short());
|
||||||
|
|
||||||
@@ -74,10 +73,6 @@ async fn main() -> Result<()> {
|
|||||||
let radio_config = cfg.radio.to_lora_config()?;
|
let radio_config = cfg.radio.to_lora_config()?;
|
||||||
let retry_policy = cfg.tx.to_retry_policy()?;
|
let retry_policy = cfg.tx.to_retry_policy()?;
|
||||||
|
|
||||||
// Derive passphrase keys.
|
|
||||||
let keys = PassphraseKeys::derive(&cfg.bridge.passphrase);
|
|
||||||
info!("topic: {}", hex::encode(&keys.topic_id.as_bytes()[..8]));
|
|
||||||
|
|
||||||
// Create rate limiter.
|
// Create rate limiter.
|
||||||
let cr_denom: u8 = match radio_config.cr {
|
let cr_denom: u8 = match radio_config.cr {
|
||||||
donglora_client::LoRaCodingRate::Cr4_5 => 5,
|
donglora_client::LoRaCodingRate::Cr4_5 => 5,
|
||||||
@@ -94,8 +89,9 @@ async fn main() -> Result<()> {
|
|||||||
);
|
);
|
||||||
info!("rate limiter: {:.1} pps", rate_limiter.rate_pps());
|
info!("rate limiter: {:.1} pps", rate_limiter.rate_pps());
|
||||||
|
|
||||||
// Create gossip network.
|
// Create gossip network + DHT rendezvous.
|
||||||
let (swarm, gossip_event_rx, gossip_frame_tx) = gossip::Gossip::new(secret_key, &keys).await?;
|
let (swarm, gossip_event_rx, gossip_frame_tx) =
|
||||||
|
gossip::Gossip::new(secret_key, &cfg.bridge.passphrase).await?;
|
||||||
let swarm = Arc::new(swarm);
|
let swarm = Arc::new(swarm);
|
||||||
|
|
||||||
// Start radio.
|
// Start radio.
|
||||||
@@ -178,13 +174,18 @@ async fn main() -> Result<()> {
|
|||||||
} else {
|
} else {
|
||||||
let terminal = tui::run(config_watch_rx, swarm.clone(), stats, log_rx, cancel.clone(), start_time);
|
let terminal = tui::run(config_watch_rx, swarm.clone(), stats, log_rx, cancel.clone(), start_time);
|
||||||
|
|
||||||
cancel.cancel();
|
// Restore the terminal first so the user's shell reappears the moment
|
||||||
swarm.shutdown().await;
|
// they hit `q`. iroh-gossip-rendezvous needs no synchronous DHT work
|
||||||
info!("donglora-bridge stopped");
|
// on shutdown — entries age out — so the only remaining wait is
|
||||||
|
// `endpoint.close()` flushing in-flight connections. Bound it.
|
||||||
if let Ok(mut term) = terminal {
|
if let Ok(mut term) = terminal {
|
||||||
tui::restore_terminal(&mut term);
|
tui::restore_terminal(&mut term);
|
||||||
}
|
}
|
||||||
|
cancel.cancel();
|
||||||
|
if tokio::time::timeout(Duration::from_secs(2), swarm.shutdown()).await.is_err() {
|
||||||
|
warn!("graceful shutdown timed out — exiting anyway");
|
||||||
|
}
|
||||||
|
info!("donglora-bridge stopped");
|
||||||
|
|
||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|||||||
+2
-2
@@ -144,7 +144,7 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn gossip_frame_roundtrip() {
|
fn gossip_frame_roundtrip() {
|
||||||
let key = iroh::SecretKey::generate(&mut rand::rng());
|
let key = iroh::SecretKey::generate();
|
||||||
let frame = GossipFrame::new(&key.public(), -85, 8, vec![1, 2, 3, 4, 5]);
|
let frame = GossipFrame::new(&key.public(), -85, 8, vec![1, 2, 3, 4, 5]);
|
||||||
let encoded = frame.encode();
|
let encoded = frame.encode();
|
||||||
let decoded = GossipFrame::decode(&encoded).expect("decode should succeed");
|
let decoded = GossipFrame::decode(&encoded).expect("decode should succeed");
|
||||||
@@ -235,7 +235,7 @@ mod tests {
|
|||||||
snr in -32i16..=32i16,
|
snr in -32i16..=32i16,
|
||||||
payload in proptest::collection::vec(any::<u8>(), 0..256),
|
payload in proptest::collection::vec(any::<u8>(), 0..256),
|
||||||
) {
|
) {
|
||||||
let key = iroh::SecretKey::generate(&mut rand::rng());
|
let key = iroh::SecretKey::generate();
|
||||||
let frame = GossipFrame::new(&key.public(), rssi, snr, payload.clone());
|
let frame = GossipFrame::new(&key.public(), rssi, snr, payload.clone());
|
||||||
let encoded = frame.encode();
|
let encoded = frame.encode();
|
||||||
let decoded = GossipFrame::decode(&encoded).unwrap();
|
let decoded = GossipFrame::decode(&encoded).unwrap();
|
||||||
|
|||||||
+11
-34
@@ -17,11 +17,11 @@ use ratatui::layout::{Alignment, Constraint, Direction, Layout, Rect};
|
|||||||
use ratatui::style::{Color, Modifier, Style};
|
use ratatui::style::{Color, Modifier, Style};
|
||||||
use ratatui::symbols;
|
use ratatui::symbols;
|
||||||
use ratatui::text::{Line, Span};
|
use ratatui::text::{Line, Span};
|
||||||
use ratatui::widgets::{Block, Borders, Clear, Paragraph, Sparkline};
|
use ratatui::widgets::{Block, Borders, Paragraph, Sparkline};
|
||||||
use tokio::sync::mpsc;
|
use tokio::sync::mpsc;
|
||||||
use tokio_util::sync::CancellationToken;
|
use tokio_util::sync::CancellationToken;
|
||||||
|
|
||||||
use crate::gossip::Gossip;
|
use crate::gossip::{DhtStatus, Gossip, RendezvousState};
|
||||||
use crate::radio::ConfigSource;
|
use crate::radio::ConfigSource;
|
||||||
use crate::router::{
|
use crate::router::{
|
||||||
PacketAction, PacketDirection, PacketLogEntry, RadioConfigInfo, Stats, StatsSnapshot, TxRetryState,
|
PacketAction, PacketDirection, PacketLogEntry, RadioConfigInfo, Stats, StatsSnapshot, TxRetryState,
|
||||||
@@ -173,7 +173,6 @@ pub fn run(
|
|||||||
&mut series,
|
&mut series,
|
||||||
);
|
);
|
||||||
|
|
||||||
terminal.draw(|frame| draw_shutdown_overlay(frame, frame.area()))?;
|
|
||||||
Ok(terminal)
|
Ok(terminal)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -210,7 +209,7 @@ fn tui_loop(
|
|||||||
let snap = stats.snapshot();
|
let snap = stats.snapshot();
|
||||||
let log_slice = log_entries.make_contiguous();
|
let log_slice = log_entries.make_contiguous();
|
||||||
series.update(&snap, log_slice);
|
series.update(&snap, log_slice);
|
||||||
let swarm_state = gossip.swarm_state();
|
let swarm_state = gossip.state();
|
||||||
let config_info = config_watch.borrow().clone();
|
let config_info = config_watch.borrow().clone();
|
||||||
terminal.draw(|frame| {
|
terminal.draw(|frame| {
|
||||||
draw_ui(frame, frame.area(), &config_info, &snap, &swarm_state, log_slice, start_time, series);
|
draw_ui(frame, frame.area(), &config_info, &snap, &swarm_state, log_slice, start_time, series);
|
||||||
@@ -241,7 +240,7 @@ fn draw_ui(
|
|||||||
area: Rect,
|
area: Rect,
|
||||||
config_info: &RadioConfigInfo,
|
config_info: &RadioConfigInfo,
|
||||||
stats: &StatsSnapshot,
|
stats: &StatsSnapshot,
|
||||||
swarm_state: &crate::gossip::SwarmState,
|
swarm_state: &RendezvousState,
|
||||||
log_entries: &[PacketLogEntry],
|
log_entries: &[PacketLogEntry],
|
||||||
start_time: Instant,
|
start_time: Instant,
|
||||||
series: &TimeSeries,
|
series: &TimeSeries,
|
||||||
@@ -316,19 +315,20 @@ fn status_pill(label: &str, ok: bool) -> Span<'static> {
|
|||||||
|
|
||||||
// ── Network panel (was Swarm) ────────────────────────────────────
|
// ── Network panel (was Swarm) ────────────────────────────────────
|
||||||
|
|
||||||
fn draw_network(frame: &mut ratatui::Frame, area: Rect, state: &crate::gossip::SwarmState) {
|
fn draw_network(frame: &mut ratatui::Frame, area: Rect, state: &RendezvousState) {
|
||||||
let block = panel_block(" Network ");
|
let block = panel_block(" Network ");
|
||||||
|
|
||||||
let dht_color = match state.dht_status {
|
let dht_color = match state.dht_status {
|
||||||
crate::gossip::DhtStatus::Ready => Color::Green,
|
DhtStatus::Ready => Color::Green,
|
||||||
crate::gossip::DhtStatus::Bootstrapping => C_RATE_DROP,
|
DhtStatus::Bootstrapping => C_RATE_DROP,
|
||||||
crate::gossip::DhtStatus::PublishFailed => C_QUEUE_DROP,
|
DhtStatus::PublishFailing => C_QUEUE_DROP,
|
||||||
};
|
};
|
||||||
let neighbor_color = if state.neighbor_count > 0 { Color::Green } else { C_RATE_DROP };
|
let neighbor_color = if state.neighbor_count > 0 { Color::Green } else { C_RATE_DROP };
|
||||||
let last_pub = state.last_dht_publish.map_or_else(|| "-".into(), |t| format!("{}s ago", t.elapsed().as_secs()));
|
let last_pub = state.last_publish.map_or_else(|| "-".into(), |t| format!("{}s ago", t.elapsed().as_secs()));
|
||||||
|
let topic_hash = hex::encode(&state.topic_id.as_bytes()[..8]);
|
||||||
|
|
||||||
let lines = vec![
|
let lines = vec![
|
||||||
kv_line("ID", &state.topic_hash, C_ACCENT),
|
kv_line("ID", &topic_hash, C_ACCENT),
|
||||||
kv_line("Peers", &format!("{}", state.neighbor_count), neighbor_color),
|
kv_line("Peers", &format!("{}", state.neighbor_count), neighbor_color),
|
||||||
kv_line("DHT", &format!("{}", state.dht_status), dht_color),
|
kv_line("DHT", &format!("{}", state.dht_status), dht_color),
|
||||||
kv_line("Published", &last_pub, C_LABEL),
|
kv_line("Published", &last_pub, C_LABEL),
|
||||||
@@ -630,29 +630,6 @@ fn pad_right_align(data: &[u64], width: usize) -> Vec<u64> {
|
|||||||
padded
|
padded
|
||||||
}
|
}
|
||||||
|
|
||||||
// ── Shutdown overlay ─────────────────────────────────────────────
|
|
||||||
|
|
||||||
fn draw_shutdown_overlay(frame: &mut ratatui::Frame, area: Rect) {
|
|
||||||
let dim = Paragraph::new("").style(Style::default().bg(Color::Black));
|
|
||||||
frame.render_widget(dim, area);
|
|
||||||
|
|
||||||
let w = 36u16;
|
|
||||||
let h = 3u16;
|
|
||||||
let x = area.x + area.width.saturating_sub(w) / 2;
|
|
||||||
let y = area.y + area.height.saturating_sub(h) / 2;
|
|
||||||
let popup = Rect::new(x, y, w.min(area.width), h.min(area.height));
|
|
||||||
|
|
||||||
frame.render_widget(Clear, popup);
|
|
||||||
let block = Block::default().borders(Borders::ALL).border_style(Style::default().fg(C_BORDER));
|
|
||||||
let text = Paragraph::new(Line::from(vec![
|
|
||||||
Span::styled(" Shutting down", Style::default().fg(C_RATE_DROP)),
|
|
||||||
Span::styled(" — updating DHT...", Style::default().fg(C_LABEL)),
|
|
||||||
]))
|
|
||||||
.alignment(Alignment::Center)
|
|
||||||
.block(block);
|
|
||||||
frame.render_widget(text, popup);
|
|
||||||
}
|
|
||||||
|
|
||||||
// ── Shared helpers ───────────────────────────────────────────────
|
// ── Shared helpers ───────────────────────────────────────────────
|
||||||
|
|
||||||
fn panel_block(title: &str) -> Block<'static> {
|
fn panel_block(title: &str) -> Block<'static> {
|
||||||
|
|||||||
Reference in new issue
Block a user