Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
57f21a0edf | ||
|
|
a30d9d5dbf | ||
|
|
2a6e6401ad | ||
|
|
a0a5922389 | ||
|
|
2c93c1c24f | ||
|
|
5564af02f9 | ||
|
|
ae29d1fea2 | ||
|
|
7ff7766ede |
@@ -6,3 +6,8 @@
|
||||
/packaging/peerspeak/
|
||||
/packaging/*.pkg.tar.*
|
||||
/packaging/*.log
|
||||
|
||||
# Windows installer build artifacts (the staged exe + compiled setup.exe);
|
||||
# the .iss script and .ico are the tracked sources.
|
||||
/packaging/windows/peerspeak.exe
|
||||
/packaging/windows/output/
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
# Maintainer: mollusk <jitty+lc1iz0dc@protonmail.com>
|
||||
#
|
||||
# Test-pack split package: ONE `makepkg -si` builds + installs BOTH peerspeak
|
||||
# (voice chat) and pixelpass (screen sharing) from the public gitbutter repos
|
||||
# over https. pixelpass lands on /usr/bin so peerspeak's screen-share button
|
||||
# finds it. Shared version string is derived from peerspeak's git.
|
||||
#
|
||||
# Clone this repo and build from here:
|
||||
# git clone https://gitbutter.xyz/mollusk/peerspeak.git
|
||||
# cd peerspeak/packaging/test-pack
|
||||
# makepkg -si
|
||||
pkgbase=peerspeak-git
|
||||
pkgname=('peerspeak-git' 'pixelpass')
|
||||
pkgver=0.1.0
|
||||
pkgrel=1
|
||||
arch=('x86_64')
|
||||
url="https://gitbutter.xyz/mollusk/peerspeak"
|
||||
license=('custom' 'MIT' 'Apache-2.0' 'OFL-1.1')
|
||||
makedepends=('git' 'cargo' 'pkgconf')
|
||||
options=('!lto' '!debug')
|
||||
source=("peerspeak::git+https://gitbutter.xyz/mollusk/peerspeak.git"
|
||||
"pixelpass::git+https://gitbutter.xyz/mollusk/pixelpass.git#branch=main")
|
||||
sha256sums=('SKIP'
|
||||
'SKIP')
|
||||
|
||||
pkgver() {
|
||||
cd "$srcdir/peerspeak"
|
||||
# Shared across both split packages. 0.1.0.r<commits>.g<short-sha>.
|
||||
printf '%s.r%s.g%s' \
|
||||
"$(awk -F'\"' '/^version =/{print $2; exit}' Cargo.toml)" \
|
||||
"$(git rev-list --count HEAD)" \
|
||||
"$(git rev-parse --short HEAD)"
|
||||
}
|
||||
|
||||
prepare() {
|
||||
# Vendor deps up front so build() can run --frozen (no surprise network).
|
||||
export CARGO_HOME="$srcdir/cargo-home"
|
||||
local host; host="$(rustc -vV | sed -n 's/host: //p')"
|
||||
cd "$srcdir/peerspeak"; cargo fetch --locked --target "$host"
|
||||
cd "$srcdir/pixelpass"; cargo fetch --locked --target "$host"
|
||||
}
|
||||
|
||||
build() {
|
||||
export CARGO_HOME="$srcdir/cargo-home"
|
||||
export RUSTUP_TOOLCHAIN=stable
|
||||
export CARGO_TARGET_DIR=target
|
||||
|
||||
cd "$srcdir/peerspeak"
|
||||
cargo build --frozen --release --bin peerspeak
|
||||
|
||||
cd "$srcdir/pixelpass"
|
||||
# --features gui so the .desktop launcher (pixelpass --gui) works.
|
||||
cargo build --frozen --release --features gui
|
||||
}
|
||||
|
||||
check() {
|
||||
export CARGO_HOME="$srcdir/cargo-home"
|
||||
export RUSTUP_TOOLCHAIN=stable
|
||||
# peerspeak library unit tests only — its integration suites bind real
|
||||
# iroh/QUIC endpoints and fail in a sandboxed/offline build environment.
|
||||
cd "$srcdir/peerspeak"
|
||||
cargo test --frozen --release --lib
|
||||
}
|
||||
|
||||
package_peerspeak-git() {
|
||||
pkgdesc="Decentralized peer-to-peer voice chat (Rust/iroh/PipeWire/Opus/iced)"
|
||||
depends=('pipewire' 'opus')
|
||||
optdepends=('pixelpass: screen sharing inside a room'
|
||||
'mpv: screen-share viewer (vlc is used as a fallback)')
|
||||
provides=('peerspeak')
|
||||
conflicts=('peerspeak')
|
||||
license=('custom')
|
||||
|
||||
cd "$srcdir/peerspeak"
|
||||
install -Dm755 "target/release/peerspeak" "$pkgdir/usr/bin/peerspeak"
|
||||
install -Dm644 "packaging/peerspeak.desktop" \
|
||||
"$pkgdir/usr/share/applications/peerspeak.desktop"
|
||||
|
||||
# Hicolor icon theme (scalable SVG + the rendered raster sizes).
|
||||
install -Dm644 "assets/icons/peerspeak.svg" \
|
||||
"$pkgdir/usr/share/icons/hicolor/scalable/apps/peerspeak.svg"
|
||||
local s
|
||||
for s in 16 24 32 48 64 128 256 512; do
|
||||
install -Dm644 "assets/icons/peerspeak-$s.png" \
|
||||
"$pkgdir/usr/share/icons/hicolor/${s}x${s}/apps/peerspeak.png"
|
||||
done
|
||||
}
|
||||
|
||||
package_pixelpass() {
|
||||
pkgdesc='P2P screen sharing over iroh — no port forwarding, no signup'
|
||||
depends=('gstreamer' 'gst-plugins-base' 'gst-plugins-good' 'gst-plugins-bad'
|
||||
'gst-libav' 'gst-plugin-va' 'libpulse' 'hicolor-icon-theme'
|
||||
'libglvnd' 'libxkbcommon' 'wayland')
|
||||
optdepends=('mpv: recommended stream viewer (the GUI launches mpv)'
|
||||
'vlc: alternative stream viewer'
|
||||
'gst-plugins-ugly: software x264 encoding for `pixelpass --no-hwencode`'
|
||||
'gst-plugin-pipewire: screen capture on Wayland sessions'
|
||||
'xorg-xwininfo: share a single window on X11 (`pixelpass --window`)')
|
||||
license=('MIT' 'Apache-2.0' 'OFL-1.1')
|
||||
|
||||
cd "$srcdir/pixelpass"
|
||||
install -Dm0755 "target/release/pixelpass" "$pkgdir/usr/bin/pixelpass"
|
||||
install -Dm0644 assets/pixelpass.desktop \
|
||||
"$pkgdir/usr/share/applications/pixelpass.desktop"
|
||||
install -Dm0644 assets/pixelpass.svg \
|
||||
"$pkgdir/usr/share/icons/hicolor/scalable/apps/pixelpass.svg"
|
||||
install -Dm0644 README.md "$pkgdir/usr/share/doc/pixelpass/README.md"
|
||||
install -Dm0644 LICENSE-MIT "$pkgdir/usr/share/licenses/pixelpass/LICENSE-MIT"
|
||||
install -Dm0644 LICENSE-APACHE "$pkgdir/usr/share/licenses/pixelpass/LICENSE-APACHE"
|
||||
install -Dm0644 assets/NotoSans-OFL.txt \
|
||||
"$pkgdir/usr/share/licenses/pixelpass/NotoSans-OFL.txt"
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
# PeerSpeak + PixelPass — CachyOS/Arch test pack
|
||||
|
||||
A single **split PKGBUILD** that builds the latest code from the public gitbutter
|
||||
repos and installs **both** programs at once:
|
||||
|
||||
- `peerspeak` — decentralized P2P voice chat
|
||||
- `pixelpass` — P2P screen sharing (peerspeak launches it for the screen-share button)
|
||||
|
||||
## Build & install (one command)
|
||||
|
||||
```sh
|
||||
git clone https://gitbutter.xyz/mollusk/peerspeak.git
|
||||
cd peerspeak/packaging/test-pack
|
||||
makepkg -si
|
||||
```
|
||||
|
||||
`makepkg -si` auto-installs every dependency via pacman before building —
|
||||
including the Rust toolchain itself (the `cargo` makedepend is provided by the
|
||||
`rust` package), `git`, `pkgconf`, pipewire + opus for peerspeak, and the
|
||||
gstreamer/VA-API stack for pixelpass. The only prerequisite is the `base-devel`
|
||||
group (which provides `makepkg`). If you already use `rustup`, that satisfies the
|
||||
`cargo` makedepend and the `rust` package won't be pulled in — no conflict.
|
||||
|
||||
When it finishes you'll have `peerspeak` and `pixelpass` on your PATH at
|
||||
`/usr/bin`. To rebuild later with fresh upstream code, re-run `makepkg -si`; the
|
||||
git sources re-pull `main` and the version bumps automatically.
|
||||
|
||||
> Skip the test step with `makepkg -si --nocheck` for a faster build.
|
||||
|
||||
## Running the cross-internet test
|
||||
|
||||
1. Launch `peerspeak` on both machines.
|
||||
2. One person **creates** a room and shares the room code/ticket with the other.
|
||||
3. The other **joins** with that code.
|
||||
4. iroh does NAT hole-punching automatically; if a direct path can't be made it
|
||||
falls back to a public n0 relay — **no port forwarding required**.
|
||||
5. Allow the app through any local firewall if prompted (outbound UDP / QUIC;
|
||||
nothing needs to be opened inbound for relay mode).
|
||||
|
||||
### What we're smoke-testing
|
||||
- Two real humans, two networks, over the internet.
|
||||
- Mic capture + remote playback both directions, no crackle/dropouts.
|
||||
- Mute / deafen, push-to-talk.
|
||||
- Text chat in-room.
|
||||
- Avatars (presets + custom upload) show up on the other side.
|
||||
- Screen share: click the screen-share control → it launches `pixelpass`; the
|
||||
viewer opens in `mpv` on the receiving side.
|
||||
- Notification chimes (join/leave/etc.).
|
||||
- Leave / rejoin cleanly.
|
||||
|
||||
If anything misbehaves, grab the log path peerspeak prints on startup and the
|
||||
exact repro steps.
|
||||
@@ -0,0 +1,82 @@
|
||||
# PeerSpeak — how to install and join a call (Windows)
|
||||
|
||||
PeerSpeak is a little voice-chat app — like a private phone call over the
|
||||
internet, with no account, no signup, and no company in the middle. You install
|
||||
it once, then you and I connect directly to each other.
|
||||
|
||||
---
|
||||
|
||||
## 1. Install it
|
||||
|
||||
1. Double-click **`peerspeak-0.2.0-setup.exe`** (the file I sent you).
|
||||
|
||||
2. **Windows will probably show a blue "Windows protected your PC" warning.**
|
||||
This is normal — it shows up for any app that isn't from a big company with a
|
||||
paid certificate. It is **not** a virus warning.
|
||||
- Click **More info**
|
||||
- Then click **Run anyway**
|
||||
|
||||
3. Windows will ask *"Do you want to allow this app to make changes?"* — click
|
||||
**Yes**.
|
||||
|
||||
4. The setup window opens. Just keep clicking **Next**. Two checkboxes you'll
|
||||
see along the way:
|
||||
- **"Allow PeerSpeak through Windows Firewall"** — leave this **checked**
|
||||
(it lets the call connect without interruptions).
|
||||
- **"Create a desktop shortcut"** — check it if you'd like an icon on your
|
||||
desktop.
|
||||
|
||||
5. Click **Install**, then **Finish**. PeerSpeak opens.
|
||||
|
||||
That's it — it's installed. You can find it again any time from the **Start
|
||||
menu** (search "PeerSpeak").
|
||||
|
||||
---
|
||||
|
||||
## 2. Get on a call with me
|
||||
|
||||
PeerSpeak connects two people using a **room ticket** — a long code that acts
|
||||
like a one-time phone number for a specific call.
|
||||
|
||||
**The simple way (I host):**
|
||||
|
||||
1. I'll create a room and send you a **ticket** (a long jumble of letters and
|
||||
numbers).
|
||||
2. Copy the whole ticket I sent you.
|
||||
3. In PeerSpeak, paste it into the **"Join Room"** box near the bottom and press
|
||||
**Join**.
|
||||
4. You're in — you should see both our names listed, and we can talk.
|
||||
|
||||
**If you want to host instead:**
|
||||
|
||||
1. Type a room name and click **Create New Room**.
|
||||
2. PeerSpeak gives you a **ticket** — click **Copy Ticket** and send it to me.
|
||||
3. I paste it on my end and join you.
|
||||
|
||||
Either way works the same; it just depends on who makes the room.
|
||||
|
||||
---
|
||||
|
||||
## 3. While you're on a call
|
||||
|
||||
- **Your microphone** is on by default. There's a **mute** button if you need
|
||||
it.
|
||||
- The first time, Windows might ask for permission to use your **microphone** —
|
||||
click **Yes / Allow**.
|
||||
- If you can't hear me or I can't hear you, open **Settings** (top right) and
|
||||
check that the right **microphone** and **speakers/headphones** are selected.
|
||||
- To hang up, click **Leave Room**.
|
||||
|
||||
---
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
- **"I don't hear anything."** Open Settings and pick the correct microphone and
|
||||
output device. Headphones are best — they prevent echo.
|
||||
- **"It won't connect."** Make sure you pasted the *entire* ticket (they're
|
||||
long and easy to cut off). If it still won't connect, we may just need a fresh
|
||||
ticket — they're meant to be used right away.
|
||||
- **The blue warning again.** Same as install: **More info → Run anyway**. It's
|
||||
the unsigned-app warning, not malware.
|
||||
|
||||
Any trouble, just message me and we'll sort it out.
|
||||
@@ -0,0 +1,72 @@
|
||||
# PeerSpeak — Windows installer
|
||||
|
||||
This directory builds a Windows setup installer for PeerSpeak using
|
||||
[Inno Setup](https://jrsoftware.org/isinfo.php).
|
||||
|
||||
PeerSpeak ships as a **single self-contained `peerspeak.exe`** — the GUI icon,
|
||||
notification chimes, and avatar presets are all embedded in the binary
|
||||
(`include_bytes!`), and the executable is statically linked against the GNU
|
||||
runtime, so there are no extra DLLs to bundle. The installer payload is just the
|
||||
`.exe` plus an `.ico` for the Start-menu / desktop shortcuts.
|
||||
|
||||
## Files
|
||||
|
||||
| File | Tracked | Purpose |
|
||||
|------|---------|---------|
|
||||
| `peerspeak.iss` | yes | Inno Setup script |
|
||||
| `peerspeak.ico` | yes | multi-resolution app icon (from `assets/icons/*.png`) |
|
||||
| `README.md` | yes | this file |
|
||||
| `peerspeak.exe` | no (gitignored) | staged build artifact, copied from `target/x86_64-pc-windows-gnu/release/` |
|
||||
| `output/peerspeak-<ver>-setup.exe` | no (gitignored) | the compiled installer |
|
||||
|
||||
## Build steps
|
||||
|
||||
1. **Cross-compile the Windows binary** (from the repo root, inside the
|
||||
`peerspeak-win` archlinux distrobox):
|
||||
|
||||
```sh
|
||||
RUSTC_BOOTSTRAP=1 ./win-cross-build.sh -Z build-std=std,panic_abort
|
||||
```
|
||||
|
||||
This needs the `rust-src` component and the `x86_64-pc-windows-gnu` target
|
||||
installed in that toolchain. The result is a statically-linked,
|
||||
GUI-subsystem `.exe` (no stray console window).
|
||||
|
||||
2. **Stage the binary** next to the script:
|
||||
|
||||
```sh
|
||||
cp target/x86_64-pc-windows-gnu/release/peerspeak.exe packaging/windows/
|
||||
```
|
||||
|
||||
3. **Regenerate the icon** if the source PNGs changed:
|
||||
|
||||
```sh
|
||||
magick assets/icons/peerspeak-16.png assets/icons/peerspeak-24.png \
|
||||
assets/icons/peerspeak-32.png assets/icons/peerspeak-48.png \
|
||||
assets/icons/peerspeak-64.png assets/icons/peerspeak-128.png \
|
||||
assets/icons/peerspeak-256.png packaging/windows/peerspeak.ico
|
||||
```
|
||||
|
||||
4. **Compile the installer** with Inno Setup. On Linux this runs under Wine:
|
||||
|
||||
```sh
|
||||
cd packaging/windows
|
||||
wine ~/.wine/drive_c/InnoSetup6/ISCC.exe peerspeak.iss
|
||||
```
|
||||
|
||||
The installer lands at `output/peerspeak-<version>-setup.exe`.
|
||||
|
||||
## What the installer does
|
||||
|
||||
- Installs `peerspeak.exe` to `Program Files\PeerSpeak` (requires admin / one
|
||||
UAC prompt).
|
||||
- Creates a Start-menu shortcut, with an optional desktop shortcut.
|
||||
- Optionally adds a Windows Firewall allow-rule for PeerSpeak (recommended —
|
||||
iroh uses UDP hole-punching, so this avoids a mid-call firewall prompt). The
|
||||
rule is removed on uninstall.
|
||||
- Provides a standard uninstaller.
|
||||
|
||||
> **Note:** the installer and the binary are **not code-signed**, so Windows
|
||||
> SmartScreen will show an "unknown publisher" warning on first run. The user
|
||||
> clicks *More info → Run anyway*. Removing this warning requires a paid
|
||||
> code-signing certificate.
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 364 KiB |
@@ -0,0 +1,63 @@
|
||||
; Inno Setup script for PeerSpeak (Windows installer).
|
||||
;
|
||||
; PeerSpeak is a single self-contained binary: the GUI icon, notification
|
||||
; chimes, and avatar presets are all embedded in the .exe (include_bytes!),
|
||||
; so the only payload here is peerspeak.exe plus an .ico for the shortcuts.
|
||||
;
|
||||
; Build (under Wine on Linux, or native Windows):
|
||||
; wine "C:\Program Files (x86)\Inno Setup 6\ISCC.exe" peerspeak.iss
|
||||
; Output lands in .\output\peerspeak-<version>-setup.exe
|
||||
;
|
||||
; The peerspeak.exe is cross-compiled with win-cross-build.sh
|
||||
; (x86_64-pc-windows-gnu, statically linked -- no extra DLLs needed).
|
||||
|
||||
#define MyAppName "PeerSpeak"
|
||||
#define MyAppVersion "0.2.0"
|
||||
#define MyAppPublisher "mollusk"
|
||||
#define MyAppExeName "peerspeak.exe"
|
||||
|
||||
[Setup]
|
||||
; A stable AppId keeps upgrades/uninstall tracking consistent across versions.
|
||||
AppId={{2754D6C1-C8A4-4B13-9824-2D303439739D}
|
||||
AppName={#MyAppName}
|
||||
AppVersion={#MyAppVersion}
|
||||
AppVerName={#MyAppName} {#MyAppVersion}
|
||||
AppPublisher={#MyAppPublisher}
|
||||
DefaultDirName={autopf}\{#MyAppName}
|
||||
DefaultGroupName={#MyAppName}
|
||||
DisableProgramGroupPage=yes
|
||||
UninstallDisplayIcon={app}\{#MyAppExeName}
|
||||
SetupIconFile=peerspeak.ico
|
||||
Compression=lzma2/max
|
||||
SolidCompression=yes
|
||||
WizardStyle=modern
|
||||
OutputDir=output
|
||||
OutputBaseFilename=peerspeak-{#MyAppVersion}-setup
|
||||
; Program Files install + firewall rule both need elevation.
|
||||
PrivilegesRequired=admin
|
||||
ArchitecturesAllowed=x64compatible
|
||||
ArchitecturesInstallIn64BitMode=x64compatible
|
||||
|
||||
[Languages]
|
||||
Name: "english"; MessagesFile: "compiler:Default.isl"
|
||||
|
||||
[Tasks]
|
||||
Name: "desktopicon"; Description: "{cm:CreateDesktopIcon}"; GroupDescription: "{cm:AdditionalIcons}"; Flags: unchecked
|
||||
Name: "firewall"; Description: "Allow PeerSpeak through Windows Firewall (recommended for voice calls)"; GroupDescription: "Network:"
|
||||
|
||||
[Files]
|
||||
Source: "peerspeak.exe"; DestDir: "{app}"; Flags: ignoreversion
|
||||
Source: "peerspeak.ico"; DestDir: "{app}"; Flags: ignoreversion
|
||||
|
||||
[Icons]
|
||||
Name: "{group}\{#MyAppName}"; Filename: "{app}\{#MyAppExeName}"; IconFilename: "{app}\peerspeak.ico"
|
||||
Name: "{group}\{cm:UninstallProgram,{#MyAppName}}"; Filename: "{uninstallexe}"
|
||||
Name: "{autodesktop}\{#MyAppName}"; Filename: "{app}\{#MyAppExeName}"; IconFilename: "{app}\peerspeak.ico"; Tasks: desktopicon
|
||||
|
||||
[Run]
|
||||
; iroh uses UDP hole-punching; pre-authorizing avoids a mid-call firewall prompt.
|
||||
Filename: "{sys}\netsh.exe"; Parameters: "advfirewall firewall add rule name=""PeerSpeak"" dir=in action=allow program=""{app}\{#MyAppExeName}"" enable=yes profile=any"; Flags: runhidden; Tasks: firewall
|
||||
Filename: "{app}\{#MyAppExeName}"; Description: "{cm:LaunchProgram,{#MyAppName}}"; Flags: nowait postinstall skipifsilent
|
||||
|
||||
[UninstallRun]
|
||||
Filename: "{sys}\netsh.exe"; Parameters: "advfirewall firewall delete rule name=""PeerSpeak"""; Flags: runhidden; RunOnceId: "DelPeerSpeakFirewall"
|
||||
@@ -888,6 +888,10 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
|
||||
state.ever_connected.remove(&id);
|
||||
notify::play(Sound::PeerLeave, state.config.custom_sound_peer_leave.as_deref());
|
||||
}
|
||||
// Core-only recovery phase: presentation for this state lands in
|
||||
// the separate UI follow-up. In particular, do not play the
|
||||
// terminal ReconnectFailed chime here.
|
||||
UiEvent::PeerRecoveryStarted { .. } => {}
|
||||
UiEvent::PeerConnectionFailed { id } => {
|
||||
state.peers.remove(&id);
|
||||
state.audio_levels.remove(&id);
|
||||
|
||||
@@ -85,6 +85,9 @@ pub enum UiEvent {
|
||||
RoomLeft,
|
||||
PeerJoined { id: EndpointId, state: PeerState },
|
||||
PeerLeft { id: EndpointId },
|
||||
/// The fixed reconnect grace expired and bounded background gossip recovery
|
||||
/// has started. This is non-terminal and must not play the failure chime.
|
||||
PeerRecoveryStarted { id: EndpointId },
|
||||
PeerConnectionFailed { id: EndpointId },
|
||||
PeerUpdated { id: EndpointId, state: PeerState },
|
||||
/// Audio link to a peer is being (re)established — show a connecting state.
|
||||
|
||||
+144
-26
@@ -1,5 +1,6 @@
|
||||
pub mod messages;
|
||||
pub mod jitter;
|
||||
mod recovery;
|
||||
|
||||
use crate::audio::{AudioBackend, PlatformAudioBackend};
|
||||
use crate::audio::eq::{Eq, EqSettings};
|
||||
@@ -11,6 +12,7 @@ use crate::network::{
|
||||
gossip::IrohGossipState,
|
||||
};
|
||||
use crate::core::messages::{CoreCommand, UiEvent};
|
||||
use crate::core::recovery::RecoveryCoordinator;
|
||||
|
||||
use crate::config::{NetworkMode, RecordingMode};
|
||||
use crate::presence::PresenceMode;
|
||||
@@ -102,6 +104,39 @@ type GraceTimers = Arc<std::sync::Mutex<HashMap<EndpointId, tokio::task::JoinHan
|
||||
/// Scrubbed whenever a peer is evicted or leaves so a later rejoin starts clean.
|
||||
type SeenConnected = Arc<std::sync::Mutex<HashSet<EndpointId>>>;
|
||||
|
||||
type KnownPeers =
|
||||
Arc<std::sync::Mutex<HashMap<String, HashMap<EndpointId, EndpointAddr>>>>;
|
||||
|
||||
#[derive(Clone)]
|
||||
struct RecoveryContext {
|
||||
coordinator: RecoveryCoordinator,
|
||||
room_state: Arc<IrohGossipState>,
|
||||
known_peers: KnownPeers,
|
||||
ticket: String,
|
||||
}
|
||||
|
||||
impl RecoveryContext {
|
||||
fn retained_addr(&self, peer_id: &EndpointId) -> Option<EndpointAddr> {
|
||||
self.known_peers
|
||||
.lock()
|
||||
.unwrap()
|
||||
.get(&self.ticket)
|
||||
.and_then(|peers| peers.get(peer_id))
|
||||
.cloned()
|
||||
}
|
||||
|
||||
fn cancel(&self, peer_id: EndpointId) {
|
||||
self.coordinator.cancel(peer_id);
|
||||
}
|
||||
|
||||
fn forget(&self, peer_id: EndpointId) {
|
||||
if let Some(peers) = self.known_peers.lock().unwrap().get_mut(&self.ticket) {
|
||||
peers.remove(&peer_id);
|
||||
}
|
||||
self.coordinator.cancel(peer_id);
|
||||
}
|
||||
}
|
||||
|
||||
/// Cancel and forget a peer's pending grace timer, if any. No-op if none is armed.
|
||||
fn cancel_grace_timer(timers: &GraceTimers, peer_id: &EndpointId) {
|
||||
if let Some(handle) = timers.lock().unwrap().remove(peer_id) {
|
||||
@@ -116,12 +151,17 @@ fn cancel_grace_timer(timers: &GraceTimers, peer_id: &EndpointId) {
|
||||
/// link repeatedly resetting the clock and dodging eviction forever. On firing it
|
||||
/// also scrubs the peer from `seen_connected` so a later rejoin isn't treated as a
|
||||
/// reconnect on its initial dial.
|
||||
struct GraceExpiry<'a> {
|
||||
transport: &'a Arc<IrohTransport>,
|
||||
jitter: &'a Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
|
||||
ui_tx: &'a mpsc::Sender<UiEvent>,
|
||||
recovery: Option<&'a RecoveryContext>,
|
||||
}
|
||||
|
||||
fn arm_grace_timer(
|
||||
timers: &GraceTimers,
|
||||
seen_connected: &SeenConnected,
|
||||
transport: &Arc<IrohTransport>,
|
||||
jitter: &Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
|
||||
ui_tx: &mpsc::Sender<UiEvent>,
|
||||
expiry: GraceExpiry<'_>,
|
||||
grace: Duration,
|
||||
peer_id: EndpointId,
|
||||
) {
|
||||
@@ -129,15 +169,29 @@ fn arm_grace_timer(
|
||||
if timers_guard.contains_key(&peer_id) {
|
||||
return;
|
||||
}
|
||||
let transport_evict = transport.clone();
|
||||
let jitter_evict = jitter.clone();
|
||||
let ui_evict = ui_tx.clone();
|
||||
let transport_evict = expiry.transport.clone();
|
||||
let jitter_evict = expiry.jitter.clone();
|
||||
let ui_evict = expiry.ui_tx.clone();
|
||||
let timers_evict = timers.clone();
|
||||
let seen_evict = seen_connected.clone();
|
||||
let recovery_evict = expiry.recovery.cloned();
|
||||
let handle = tokio::spawn(async move {
|
||||
tokio::time::sleep(grace).await;
|
||||
crate::log_msg(&format!("Reconnect grace expired; evicting peer {:?}", peer_id));
|
||||
crate::log_msg(&format!("Reconnect grace expired for peer {:?}", peer_id));
|
||||
|
||||
if let Some(recovery) = &recovery_evict
|
||||
&& !recovery.coordinator.begin(peer_id)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
transport_evict.remove_audio_sender(peer_id);
|
||||
if let Some(recovery) = &recovery_evict {
|
||||
// Revoke roster authority before the first await in teardown. A
|
||||
// verified Announce racing after this point is then a PeerJoined and
|
||||
// cancels recovery instead of being erased after it was accepted.
|
||||
recovery.room_state.mark_peer_disconnected(peer_id);
|
||||
}
|
||||
transport_evict.disconnect_peer(peer_id).await;
|
||||
jitter_evict.lock().await.remove(&peer_id);
|
||||
// Scrub our internal state *before* announcing the eviction, so anything
|
||||
@@ -146,7 +200,38 @@ fn arm_grace_timer(
|
||||
// reconnect.
|
||||
timers_evict.lock().unwrap().remove(&peer_id);
|
||||
seen_evict.lock().unwrap().remove(&peer_id);
|
||||
let _ = ui_evict.send(UiEvent::PeerConnectionFailed { id: peer_id }).await;
|
||||
|
||||
let Some(recovery) = recovery_evict else {
|
||||
let _ = ui_evict.send(UiEvent::PeerConnectionFailed { id: peer_id }).await;
|
||||
return;
|
||||
};
|
||||
if !recovery.coordinator.is_active(&peer_id) {
|
||||
return;
|
||||
}
|
||||
|
||||
let Some(addr) = recovery.retained_addr(&peer_id) else {
|
||||
crate::log_msg(&format!(
|
||||
"Cannot recover peer {:?}: no retained authenticated address",
|
||||
peer_id
|
||||
));
|
||||
recovery.cancel(peer_id);
|
||||
let _ = ui_evict.send(UiEvent::PeerConnectionFailed { id: peer_id }).await;
|
||||
return;
|
||||
};
|
||||
|
||||
match recovery.coordinator.activate(peer_id, addr) {
|
||||
Ok(true) => {
|
||||
let _ = ui_evict.send(UiEvent::PeerRecoveryStarted { id: peer_id }).await;
|
||||
}
|
||||
Ok(false) => {}
|
||||
Err(()) => {
|
||||
crate::log_msg(&format!(
|
||||
"Cannot recover peer {:?}: recovery coordinator unavailable",
|
||||
peer_id
|
||||
));
|
||||
let _ = ui_evict.send(UiEvent::PeerConnectionFailed { id: peer_id }).await;
|
||||
}
|
||||
}
|
||||
});
|
||||
timers_guard.insert(peer_id, handle);
|
||||
}
|
||||
@@ -309,6 +394,7 @@ pub struct ConnEventHandler {
|
||||
seen_connected: SeenConnected,
|
||||
transport: Arc<IrohTransport>,
|
||||
jitter: Arc<Mutex<HashMap<EndpointId, JitterBuffer>>>,
|
||||
recovery: Option<RecoveryContext>,
|
||||
grace: Duration,
|
||||
}
|
||||
|
||||
@@ -326,6 +412,7 @@ impl ConnEventHandler {
|
||||
seen_connected,
|
||||
transport,
|
||||
jitter,
|
||||
recovery: None,
|
||||
grace: RECONNECT_GRACE,
|
||||
}
|
||||
}
|
||||
@@ -336,6 +423,11 @@ impl ConnEventHandler {
|
||||
self
|
||||
}
|
||||
|
||||
fn with_recovery(mut self, recovery: RecoveryContext) -> Self {
|
||||
self.recovery = Some(recovery);
|
||||
self
|
||||
}
|
||||
|
||||
pub async fn handle(&self, event: ConnEvent) {
|
||||
match event {
|
||||
ConnEvent::Connecting(id) => {
|
||||
@@ -349,9 +441,12 @@ impl ConnEventHandler {
|
||||
arm_grace_timer(
|
||||
&self.grace_timers,
|
||||
&self.seen_connected,
|
||||
&self.transport,
|
||||
&self.jitter,
|
||||
&self.ui_tx,
|
||||
GraceExpiry {
|
||||
transport: &self.transport,
|
||||
jitter: &self.jitter,
|
||||
ui_tx: &self.ui_tx,
|
||||
recovery: self.recovery.as_ref(),
|
||||
},
|
||||
self.grace,
|
||||
id,
|
||||
);
|
||||
@@ -359,6 +454,15 @@ impl ConnEventHandler {
|
||||
let _ = self.ui_tx.send(UiEvent::PeerConnecting { id }).await;
|
||||
}
|
||||
ConnEvent::Connected(id) => {
|
||||
// A transport event cannot readmit a grace-expired peer. Ignore a
|
||||
// stale/racing link until authenticated gossip emits PeerJoined.
|
||||
if self
|
||||
.recovery
|
||||
.as_ref()
|
||||
.is_some_and(|recovery| recovery.coordinator.is_active(&id))
|
||||
{
|
||||
return;
|
||||
}
|
||||
// The audio link came back — the peer recovered within the grace
|
||||
// window, so cancel its eviction.
|
||||
cancel_grace_timer(&self.grace_timers, &id);
|
||||
@@ -371,6 +475,9 @@ impl ConnEventHandler {
|
||||
// until the grace timer or the slow gossip Leave.
|
||||
cancel_grace_timer(&self.grace_timers, &id);
|
||||
self.seen_connected.lock().unwrap().remove(&id);
|
||||
if let Some(recovery) = &self.recovery {
|
||||
recovery.forget(id);
|
||||
}
|
||||
self.transport.remove_audio_sender(id);
|
||||
self.transport.disconnect_peer(id).await;
|
||||
self.jitter.lock().await.remove(&id);
|
||||
@@ -387,6 +494,7 @@ struct ActiveSession {
|
||||
mixer_task: tokio::task::JoinHandle<()>,
|
||||
event_task: tokio::task::JoinHandle<()>,
|
||||
conn_event_task: tokio::task::JoinHandle<()>,
|
||||
recovery_task: tokio::task::JoinHandle<()>,
|
||||
grace_timers: GraceTimers,
|
||||
transport: Arc<IrohTransport>,
|
||||
/// Loaded PipeWire echo-cancel module (if enabled); unloads on drop.
|
||||
@@ -421,6 +529,7 @@ impl ActiveSession {
|
||||
for (_, handle) in self.grace_timers.lock().unwrap().drain() {
|
||||
handle.abort();
|
||||
}
|
||||
self.recovery_task.abort();
|
||||
crate::log_msg("Aborted tasks");
|
||||
|
||||
let audio_backend_clone = audio_backend.clone();
|
||||
@@ -730,8 +839,7 @@ async fn run_core_loop(
|
||||
// first room's peers — the old single-set version cleared them on any ticket
|
||||
// change, so an A→B→A bounce stranded the rejoiner with an empty bootstrap.
|
||||
// Inner map keyed by peer id so updates refresh the address.
|
||||
let known_peers: Arc<std::sync::Mutex<HashMap<String, HashMap<EndpointId, EndpointAddr>>>> =
|
||||
Arc::new(std::sync::Mutex::new(HashMap::new()));
|
||||
let known_peers: KnownPeers = Arc::new(std::sync::Mutex::new(HashMap::new()));
|
||||
|
||||
let audio_backend = Arc::new(PlatformAudioBackend::new());
|
||||
|
||||
@@ -1474,6 +1582,15 @@ async fn run_core_loop(
|
||||
// The ticket of the room this event loop serves, so peer add/remove
|
||||
// updates the right per-ticket bucket in `known_peers` (A8 archive).
|
||||
let ticket_events = ticket_str.clone();
|
||||
let (recovery_coordinator, recovery_task) =
|
||||
RecoveryCoordinator::spawn(room_state.clone());
|
||||
let recovery_context = RecoveryContext {
|
||||
coordinator: recovery_coordinator,
|
||||
room_state: room_state.clone(),
|
||||
known_peers: known_peers.clone(),
|
||||
ticket: ticket_str.clone(),
|
||||
};
|
||||
let recovery_events = recovery_context.clone();
|
||||
// Friends store + ui sender, so a connected peer who is a friend has
|
||||
// their saved address auto-healed (W7) — populates `last_addr` so the
|
||||
// presence scheduler can reach them later.
|
||||
@@ -1486,6 +1603,7 @@ async fn run_core_loop(
|
||||
// A (re)join means the peer is back — cancel any
|
||||
// pending reconnect grace timer before re-adding it.
|
||||
cancel_grace_timer(&grace_timers_events, &peer_id);
|
||||
recovery_events.cancel(peer_id);
|
||||
transport_events.admit_audio_sender(peer_id);
|
||||
// Establish the audio connection as soon as the peer
|
||||
// is known (the transport dedupes the full-mesh race).
|
||||
@@ -1529,15 +1647,9 @@ async fn run_core_loop(
|
||||
// Graceful leave — evict immediately.
|
||||
cancel_grace_timer(&grace_timers_events, &peer_id);
|
||||
seen_connected_events.lock().unwrap().remove(&peer_id);
|
||||
// Graceful leave: drop them as a rejoin dial target
|
||||
// for this room (a transient PeerConnectionLost
|
||||
// deliberately does NOT, so we can still re-dial a
|
||||
// peer who's still up).
|
||||
if let Some(peers) =
|
||||
known_peers_events.lock().unwrap().get_mut(&ticket_events)
|
||||
{
|
||||
peers.remove(&peer_id);
|
||||
}
|
||||
// A signed Leave cancels background recovery and
|
||||
// drops the retained target. Transient loss keeps it.
|
||||
recovery_events.forget(peer_id);
|
||||
transport_events.remove_audio_sender(peer_id);
|
||||
transport_events.disconnect_peer(peer_id).await;
|
||||
jitter_events.lock().await.remove(&peer_id);
|
||||
@@ -1551,6 +1663,7 @@ async fn run_core_loop(
|
||||
// it. Idempotent: an ordinary mute/unmute update just
|
||||
// re-records the same address.
|
||||
cancel_grace_timer(&grace_timers_events, &peer_id);
|
||||
recovery_events.cancel(peer_id);
|
||||
transport_events.admit_audio_sender(peer_id);
|
||||
transport_events.connect_peer(state.addr.clone()).await;
|
||||
// Auto-heal a friend's saved address (W7) on the
|
||||
@@ -1598,9 +1711,12 @@ async fn run_core_loop(
|
||||
arm_grace_timer(
|
||||
&grace_timers_events,
|
||||
&seen_connected_events,
|
||||
&transport_events,
|
||||
&jitter_events,
|
||||
&ui_tx_events,
|
||||
GraceExpiry {
|
||||
transport: &transport_events,
|
||||
jitter: &jitter_events,
|
||||
ui_tx: &ui_tx_events,
|
||||
recovery: Some(&recovery_events),
|
||||
},
|
||||
RECONNECT_GRACE,
|
||||
peer_id,
|
||||
);
|
||||
@@ -1624,7 +1740,8 @@ async fn run_core_loop(
|
||||
seen_connected.clone(),
|
||||
transport.clone(),
|
||||
jitter.clone(),
|
||||
);
|
||||
)
|
||||
.with_recovery(recovery_context);
|
||||
let conn_event_task = tokio::spawn(async move {
|
||||
while let Some(event) = conn_events.recv().await {
|
||||
conn_handler.handle(event).await;
|
||||
@@ -1638,6 +1755,7 @@ async fn run_core_loop(
|
||||
mixer_task,
|
||||
event_task,
|
||||
conn_event_task,
|
||||
recovery_task,
|
||||
grace_timers,
|
||||
transport: transport.clone(),
|
||||
#[cfg(target_os = "linux")]
|
||||
|
||||
@@ -0,0 +1,264 @@
|
||||
use crate::network::{RoomState, gossip::IrohGossipState};
|
||||
use iroh::{EndpointAddr, EndpointId};
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::task::JoinHandle;
|
||||
use tokio::time::Instant;
|
||||
|
||||
const RECOVERY_COMMAND_CAPACITY: usize = 64;
|
||||
const RECOVERY_DELAYS: [Duration; 7] = [
|
||||
Duration::from_secs(1),
|
||||
Duration::from_secs(2),
|
||||
Duration::from_secs(4),
|
||||
Duration::from_secs(8),
|
||||
Duration::from_secs(15),
|
||||
Duration::from_secs(30),
|
||||
Duration::from_secs(60),
|
||||
];
|
||||
|
||||
fn recovery_delay(attempt: usize) -> Duration {
|
||||
RECOVERY_DELAYS[attempt.min(RECOVERY_DELAYS.len() - 1)]
|
||||
}
|
||||
|
||||
enum RecoveryCommand {
|
||||
Start {
|
||||
peer_id: EndpointId,
|
||||
addr: EndpointAddr,
|
||||
},
|
||||
Cancel(EndpointId),
|
||||
}
|
||||
|
||||
struct RecoveryEntry {
|
||||
addr: EndpointAddr,
|
||||
attempt: usize,
|
||||
next_attempt: Instant,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
trait RecoveryRoom: Send + Sync {
|
||||
async fn rebootstrap_peers(&self, peers: Vec<EndpointAddr>) -> Result<(), String>;
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl RecoveryRoom for IrohGossipState {
|
||||
async fn rebootstrap_peers(&self, peers: Vec<EndpointAddr>) -> Result<(), String> {
|
||||
RoomState::rebootstrap_peers(self, peers)
|
||||
.await
|
||||
.map_err(|error| error.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
/// Cloneable command side of the single per-session recovery coordinator.
|
||||
/// `active` is shared with transport/event handlers so cancellation is visible
|
||||
/// immediately even while the coordinator is awaiting an in-flight gossip call.
|
||||
#[derive(Clone)]
|
||||
pub(super) struct RecoveryCoordinator {
|
||||
tx: mpsc::Sender<RecoveryCommand>,
|
||||
active: Arc<Mutex<HashSet<EndpointId>>>,
|
||||
}
|
||||
|
||||
impl RecoveryCoordinator {
|
||||
pub(super) fn spawn(room_state: Arc<IrohGossipState>) -> (Self, JoinHandle<()>) {
|
||||
Self::spawn_inner(room_state)
|
||||
}
|
||||
|
||||
fn spawn_inner(room_state: Arc<dyn RecoveryRoom>) -> (Self, JoinHandle<()>) {
|
||||
let (tx, rx) = mpsc::channel(RECOVERY_COMMAND_CAPACITY);
|
||||
let active = Arc::new(Mutex::new(HashSet::new()));
|
||||
let handle = Self {
|
||||
tx,
|
||||
active: active.clone(),
|
||||
};
|
||||
let task = tokio::spawn(run_coordinator(room_state, active, rx));
|
||||
(handle, task)
|
||||
}
|
||||
|
||||
/// Reserve one recovery slot before grace-expiry teardown begins. Returns
|
||||
/// false when the peer is already recovering, preventing duplicate work.
|
||||
pub(super) fn begin(&self, peer_id: EndpointId) -> bool {
|
||||
self.active.lock().unwrap().insert(peer_id)
|
||||
}
|
||||
|
||||
/// Activate the reserved slot with its retained authenticated address.
|
||||
/// Uses a bounded non-blocking send while holding the active-set lock so a
|
||||
/// concurrent cancellation is ordered before or after this command.
|
||||
pub(super) fn activate(&self, peer_id: EndpointId, addr: EndpointAddr) -> Result<bool, ()> {
|
||||
let mut active = self.active.lock().unwrap();
|
||||
if !active.contains(&peer_id) {
|
||||
return Ok(false);
|
||||
}
|
||||
if self
|
||||
.tx
|
||||
.try_send(RecoveryCommand::Start { peer_id, addr })
|
||||
.is_err()
|
||||
{
|
||||
active.remove(&peer_id);
|
||||
return Err(());
|
||||
}
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
pub(super) fn cancel(&self, peer_id: EndpointId) {
|
||||
self.active.lock().unwrap().remove(&peer_id);
|
||||
// Cancellation is governed by the shared active set, so it remains
|
||||
// immediate even if the bounded command queue is temporarily full.
|
||||
let _ = self.tx.try_send(RecoveryCommand::Cancel(peer_id));
|
||||
}
|
||||
|
||||
pub(super) fn is_active(&self, peer_id: &EndpointId) -> bool {
|
||||
self.active.lock().unwrap().contains(peer_id)
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_coordinator(
|
||||
room_state: Arc<dyn RecoveryRoom>,
|
||||
active: Arc<Mutex<HashSet<EndpointId>>>,
|
||||
mut rx: mpsc::Receiver<RecoveryCommand>,
|
||||
) {
|
||||
let mut entries: HashMap<EndpointId, RecoveryEntry> = HashMap::new();
|
||||
|
||||
loop {
|
||||
// The shared active set is the authoritative cancellation gate. Prune
|
||||
// here as well as on Cancel commands so a saturated command queue cannot
|
||||
// leave an inactive, past-due entry spinning the timer loop.
|
||||
let active_snapshot = active.lock().unwrap().clone();
|
||||
entries.retain(|peer_id, _| active_snapshot.contains(peer_id));
|
||||
let next_deadline = entries.values().map(|entry| entry.next_attempt).min();
|
||||
let command = match next_deadline {
|
||||
Some(deadline) => {
|
||||
tokio::select! {
|
||||
command = rx.recv() => command,
|
||||
_ = tokio::time::sleep_until(deadline) => {
|
||||
let now = Instant::now();
|
||||
let active_snapshot = active.lock().unwrap().clone();
|
||||
let due: Vec<(EndpointId, EndpointAddr)> = entries
|
||||
.iter()
|
||||
.filter(|(id, entry)| {
|
||||
entry.next_attempt <= now && active_snapshot.contains(*id)
|
||||
})
|
||||
.map(|(id, entry)| (*id, entry.addr.clone()))
|
||||
.collect();
|
||||
|
||||
if !due.is_empty() {
|
||||
let addrs = due.iter().map(|(_, addr)| addr.clone()).collect();
|
||||
if let Err(error) = room_state.rebootstrap_peers(addrs).await {
|
||||
crate::log_msg(&format!(
|
||||
"Background peer recovery attempt failed: {error}"
|
||||
));
|
||||
}
|
||||
|
||||
let scheduled_at = Instant::now();
|
||||
for (peer_id, _) in due {
|
||||
if !active.lock().unwrap().contains(&peer_id) {
|
||||
entries.remove(&peer_id);
|
||||
continue;
|
||||
}
|
||||
if let Some(entry) = entries.get_mut(&peer_id) {
|
||||
entry.next_attempt = scheduled_at + recovery_delay(entry.attempt);
|
||||
entry.attempt = entry.attempt.saturating_add(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
None => rx.recv().await,
|
||||
};
|
||||
|
||||
match command {
|
||||
Some(RecoveryCommand::Start { peer_id, addr }) => {
|
||||
if active.lock().unwrap().contains(&peer_id) {
|
||||
entries.entry(peer_id).or_insert(RecoveryEntry {
|
||||
addr,
|
||||
attempt: 0,
|
||||
next_attempt: Instant::now(),
|
||||
});
|
||||
}
|
||||
}
|
||||
Some(RecoveryCommand::Cancel(peer_id)) => {
|
||||
entries.remove(&peer_id);
|
||||
}
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use iroh::SecretKey;
|
||||
|
||||
struct RecordingRoom {
|
||||
attempts: mpsc::UnboundedSender<Vec<EndpointAddr>>,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl RecoveryRoom for RecordingRoom {
|
||||
async fn rebootstrap_peers(&self, peers: Vec<EndpointAddr>) -> Result<(), String> {
|
||||
self.attempts.send(peers).map_err(|error| error.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retry_backoff_reaches_and_stays_at_sixty_seconds() {
|
||||
let actual: Vec<u64> = (0..10)
|
||||
.map(|attempt| recovery_delay(attempt).as_secs())
|
||||
.collect();
|
||||
assert_eq!(actual, vec![1, 2, 4, 8, 15, 30, 60, 60, 60, 60]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recovery_slots_are_deduplicated_and_cancel_immediately() {
|
||||
let (tx, mut rx) = mpsc::channel(4);
|
||||
let coordinator = RecoveryCoordinator {
|
||||
tx,
|
||||
active: Arc::new(Mutex::new(HashSet::new())),
|
||||
};
|
||||
let peer_id = SecretKey::generate().public();
|
||||
|
||||
assert!(coordinator.begin(peer_id));
|
||||
assert!(
|
||||
!coordinator.begin(peer_id),
|
||||
"a peer gets only one recovery slot"
|
||||
);
|
||||
assert_eq!(
|
||||
coordinator.activate(peer_id, EndpointAddr::from(peer_id)),
|
||||
Ok(true)
|
||||
);
|
||||
assert!(matches!(
|
||||
rx.try_recv(),
|
||||
Ok(RecoveryCommand::Start { peer_id: id, .. }) if id == peer_id
|
||||
));
|
||||
|
||||
coordinator.cancel(peer_id);
|
||||
assert!(!coordinator.is_active(&peer_id));
|
||||
assert!(matches!(
|
||||
rx.try_recv(),
|
||||
Ok(RecoveryCommand::Cancel(id)) if id == peer_id
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn coordinator_attempts_rebootstrap_immediately() {
|
||||
let (attempts_tx, mut attempts_rx) = mpsc::unbounded_channel();
|
||||
let (coordinator, task) = RecoveryCoordinator::spawn_inner(Arc::new(RecordingRoom {
|
||||
attempts: attempts_tx,
|
||||
}));
|
||||
let peer_id = SecretKey::generate().public();
|
||||
let addr = EndpointAddr::from(peer_id);
|
||||
|
||||
assert!(coordinator.begin(peer_id));
|
||||
assert_eq!(coordinator.activate(peer_id, addr.clone()), Ok(true));
|
||||
let attempted = tokio::time::timeout(Duration::from_secs(1), attempts_rx.recv())
|
||||
.await
|
||||
.expect("first recovery attempt should be immediate")
|
||||
.expect("recording room remains subscribed");
|
||||
assert_eq!(attempted, vec![addr]);
|
||||
|
||||
coordinator.cancel(peer_id);
|
||||
task.abort();
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,7 @@
|
||||
// On Windows, suppress the extra console window for release GUI builds while
|
||||
// keeping it in debug builds so stderr/panics stay visible during development.
|
||||
#![cfg_attr(not(debug_assertions), windows_subsystem = "windows")]
|
||||
|
||||
fn main() {
|
||||
if let Err(e) = peerspeak::app::run_gui() {
|
||||
eprintln!("Error running GUI: {:?}", e);
|
||||
|
||||
+52
-2
@@ -5,7 +5,7 @@ use iroh_gossip::proto::TopicId;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::sync::mpsc::Receiver;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::collections::HashMap;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use async_trait::async_trait;
|
||||
use tokio_stream::StreamExt;
|
||||
use serde::{Serialize, Deserialize};
|
||||
@@ -198,6 +198,10 @@ pub struct IrohGossipState {
|
||||
secret_key: SecretKey,
|
||||
self_state: Arc<Mutex<Option<PeerState>>>,
|
||||
peers: Arc<Mutex<HashMap<EndpointId, PeerState>>>,
|
||||
/// Previously verified peers whose live roster entry was removed by a
|
||||
/// transient disconnect. Retained only so a later authenticated `Leave`
|
||||
/// still reaches core and cancels background recovery.
|
||||
disconnected_peers: Arc<Mutex<HashSet<EndpointId>>>,
|
||||
event_tx: mpsc::Sender<RoomEvent>,
|
||||
event_rx: Mutex<Option<mpsc::Receiver<RoomEvent>>>,
|
||||
active_topic: Mutex<Option<tokio::task::JoinHandle<()>>>,
|
||||
@@ -223,6 +227,7 @@ impl IrohGossipState {
|
||||
secret_key,
|
||||
self_state: Arc::new(Mutex::new(None)),
|
||||
peers: Arc::new(Mutex::new(HashMap::new())),
|
||||
disconnected_peers: Arc::new(Mutex::new(HashSet::new())),
|
||||
event_tx,
|
||||
event_rx: Mutex::new(Some(event_rx)),
|
||||
active_topic: Mutex::new(None),
|
||||
@@ -295,6 +300,7 @@ impl RoomState for IrohGossipState {
|
||||
|
||||
let event_tx = self.event_tx.clone();
|
||||
let peers = self.peers.clone();
|
||||
let disconnected_peers = self.disconnected_peers.clone();
|
||||
let address_lookup = self.address_lookup.clone();
|
||||
let self_state_clone = self.self_state.clone();
|
||||
let gossip_sender_clone = gossip_sender.clone();
|
||||
@@ -393,6 +399,7 @@ impl RoomState for IrohGossipState {
|
||||
// peer-supplied: cap/validate once at ingest
|
||||
// so invalid offers never render a Watch button.
|
||||
state.sharing = state.sharing.and_then(crate::screenshare::sanitize_ticket);
|
||||
disconnected_peers.lock().unwrap().remove(&payload.author);
|
||||
let (is_new, state_changed) = {
|
||||
let mut peer_map = peers.lock().unwrap();
|
||||
let is_new = !peer_map.contains_key(&payload.author);
|
||||
@@ -423,7 +430,11 @@ impl RoomState for IrohGossipState {
|
||||
GossipMessage::Leave => {
|
||||
crate::log_msg(&format!("Gossip peer leave request from author={:?}", payload.author));
|
||||
let removed = peers.lock().unwrap().remove(&payload.author).is_some();
|
||||
if removed {
|
||||
let was_disconnected = disconnected_peers
|
||||
.lock()
|
||||
.unwrap()
|
||||
.remove(&payload.author);
|
||||
if removed || was_disconnected {
|
||||
let _ = event_tx.send(RoomEvent::PeerLeft(payload.author)).await;
|
||||
}
|
||||
}
|
||||
@@ -472,6 +483,7 @@ impl RoomState for IrohGossipState {
|
||||
// cached presence entry; a rejoin re-announces as new.
|
||||
let removed = peers.lock().unwrap().remove(&peer_id).is_some();
|
||||
if removed {
|
||||
disconnected_peers.lock().unwrap().insert(peer_id);
|
||||
crate::log_msg(&format!("Peer connection lost (NeighborDown): {:?}", peer_id));
|
||||
let _ = event_tx.send(RoomEvent::PeerConnectionLost(peer_id)).await;
|
||||
}
|
||||
@@ -516,6 +528,43 @@ impl RoomState for IrohGossipState {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn rebootstrap_peers(&self, peers: Vec<EndpointAddr>) -> Result<(), NetError> {
|
||||
let self_id = self._endpoint.id();
|
||||
let mut peer_ids = Vec::new();
|
||||
for addr in peers {
|
||||
if addr.id == self_id || peer_ids.contains(&addr.id) {
|
||||
continue;
|
||||
}
|
||||
self.address_lookup.add_endpoint_info(addr.clone());
|
||||
peer_ids.push(addr.id);
|
||||
}
|
||||
|
||||
if peer_ids.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Clone the sender before awaiting: active_sender is a standard mutex and
|
||||
// must never be held across an async gossip operation.
|
||||
let sender = self
|
||||
.active_sender
|
||||
.lock()
|
||||
.unwrap()
|
||||
.clone()
|
||||
.ok_or_else(|| NetError::Other("Not in a room".to_string()))?;
|
||||
|
||||
crate::log_msg(&format!("Rebootstrapping gossip peers: {:?}", peer_ids));
|
||||
sender
|
||||
.join_peers(peer_ids)
|
||||
.await
|
||||
.map_err(|e| NetError::Gossip(e.to_string()))
|
||||
}
|
||||
|
||||
fn mark_peer_disconnected(&self, peer_id: EndpointId) {
|
||||
if self.peers.lock().unwrap().remove(&peer_id).is_some() {
|
||||
self.disconnected_peers.lock().unwrap().insert(peer_id);
|
||||
}
|
||||
}
|
||||
|
||||
async fn send_chat(&self, text: String) -> Result<(), NetError> {
|
||||
let name = {
|
||||
let guard = self.self_state.lock().unwrap();
|
||||
@@ -571,6 +620,7 @@ impl RoomState for IrohGossipState {
|
||||
}
|
||||
|
||||
self.peers.lock().unwrap().clear();
|
||||
self.disconnected_peers.lock().unwrap().clear();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
+11
-1
@@ -194,6 +194,17 @@ pub trait RoomState: Send + Sync {
|
||||
/// Updates our local state (e.g. when user mutes/unmutes) and broadcasts it.
|
||||
async fn update_self_state(&self, self_state: PeerState) -> Result<(), NetError>;
|
||||
|
||||
/// Ask the active gossip topic to connect to retained peer addresses without
|
||||
/// leaving or replacing the subscription. This is a recovery primitive only:
|
||||
/// it does not add peers to the authenticated room roster. A peer becomes
|
||||
/// active only after its normal signed `Announce` is received and verified.
|
||||
async fn rebootstrap_peers(&self, peers: Vec<EndpointAddr>) -> Result<(), NetError>;
|
||||
|
||||
/// Remove a peer from the authenticated live roster before background
|
||||
/// recovery. This only revokes membership; a fresh verified `Announce` is
|
||||
/// required to add the peer again.
|
||||
fn mark_peer_disconnected(&self, peer_id: EndpointId);
|
||||
|
||||
/// Broadcasts a room text-chat message authored by us (our display name is
|
||||
/// taken from the current self-state).
|
||||
async fn send_chat(&self, text: String) -> Result<(), NetError>;
|
||||
@@ -342,4 +353,3 @@ mod tests {
|
||||
assert_eq!(original, deserialized);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+64
-4
@@ -60,6 +60,9 @@ pub enum AppTheme {
|
||||
GruvboxDark,
|
||||
SolarizedLight,
|
||||
GruvboxLight,
|
||||
AyuDark,
|
||||
AyuMirage,
|
||||
AyuLight,
|
||||
}
|
||||
|
||||
/// Build a `Color` from a packed `0xRRGGBB` literal — keeps the palette tables
|
||||
@@ -70,7 +73,7 @@ fn hex(c: u32) -> Color {
|
||||
|
||||
impl AppTheme {
|
||||
/// Every theme, in picker order.
|
||||
pub const ALL: [AppTheme; 10] = [
|
||||
pub const ALL: [AppTheme; 13] = [
|
||||
AppTheme::Mocha,
|
||||
AppTheme::Macchiato,
|
||||
AppTheme::Frappe,
|
||||
@@ -81,6 +84,9 @@ impl AppTheme {
|
||||
AppTheme::GruvboxDark,
|
||||
AppTheme::SolarizedLight,
|
||||
AppTheme::GruvboxLight,
|
||||
AppTheme::AyuDark,
|
||||
AppTheme::AyuMirage,
|
||||
AppTheme::AyuLight,
|
||||
];
|
||||
|
||||
/// Human-readable name for the picker.
|
||||
@@ -96,6 +102,9 @@ impl AppTheme {
|
||||
AppTheme::GruvboxDark => "Gruvbox Dark",
|
||||
AppTheme::SolarizedLight => "Solarized Light",
|
||||
AppTheme::GruvboxLight => "Gruvbox Light",
|
||||
AppTheme::AyuDark => "Ayu Dark",
|
||||
AppTheme::AyuMirage => "Ayu Mirage",
|
||||
AppTheme::AyuLight => "Ayu Light",
|
||||
}
|
||||
}
|
||||
|
||||
@@ -103,7 +112,10 @@ impl AppTheme {
|
||||
pub fn is_dark(self) -> bool {
|
||||
!matches!(
|
||||
self,
|
||||
AppTheme::Latte | AppTheme::SolarizedLight | AppTheme::GruvboxLight
|
||||
AppTheme::Latte
|
||||
| AppTheme::SolarizedLight
|
||||
| AppTheme::GruvboxLight
|
||||
| AppTheme::AyuLight
|
||||
)
|
||||
}
|
||||
|
||||
@@ -120,6 +132,8 @@ impl AppTheme {
|
||||
AppTheme::GruvboxDark => iced::Theme::GruvboxDark,
|
||||
AppTheme::SolarizedLight => iced::Theme::SolarizedLight,
|
||||
AppTheme::GruvboxLight => iced::Theme::GruvboxLight,
|
||||
AppTheme::AyuDark | AppTheme::AyuMirage => iced::Theme::TokyoNight,
|
||||
AppTheme::AyuLight => iced::Theme::Light,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -281,6 +295,52 @@ impl AppTheme {
|
||||
green: hex(0x79740e),
|
||||
yellow: hex(0xb57614),
|
||||
},
|
||||
AppTheme::AyuDark => Palette {
|
||||
crust: hex(0x06080a),
|
||||
mantle: hex(0x0b0e14),
|
||||
base: hex(0x0d1017),
|
||||
surface: hex(0x1c222b),
|
||||
overlay: hex(0x565b66),
|
||||
text: hex(0xbfbdb6),
|
||||
subtext: hex(0x9da1a6),
|
||||
blue: hex(0xe6b450),
|
||||
lavender: hex(0x59c2ff),
|
||||
red: hex(0xf07178),
|
||||
maroon: hex(0xff8f40),
|
||||
green: hex(0xaad94c),
|
||||
yellow: hex(0xffb454),
|
||||
},
|
||||
AppTheme::AyuMirage => Palette {
|
||||
crust: hex(0x171b24),
|
||||
mantle: hex(0x1a1f29),
|
||||
base: hex(0x1f2430),
|
||||
surface: hex(0x232834),
|
||||
overlay: hex(0x707a8c),
|
||||
text: hex(0xcccac2),
|
||||
subtext: hex(0xa6abb4),
|
||||
blue: hex(0xffcc66),
|
||||
lavender: hex(0x73d0ff),
|
||||
red: hex(0xf28779),
|
||||
maroon: hex(0xffa759),
|
||||
green: hex(0xd5ff80),
|
||||
yellow: hex(0xffd173),
|
||||
},
|
||||
// Ayu Light's canonical orange is deepened for legibility on white.
|
||||
AppTheme::AyuLight => Palette {
|
||||
crust: hex(0xe6e9ec),
|
||||
mantle: hex(0xf3f4f5),
|
||||
base: hex(0xfcfcfc),
|
||||
surface: hex(0xe8eaed),
|
||||
overlay: hex(0x8a9199),
|
||||
text: hex(0x5c6166),
|
||||
subtext: hex(0x737980),
|
||||
blue: hex(0xc7500e),
|
||||
lavender: hex(0x399ee6),
|
||||
red: hex(0xf07171),
|
||||
maroon: hex(0xfa8d3e),
|
||||
green: hex(0x86b300),
|
||||
yellow: hex(0xff9940),
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -371,11 +431,11 @@ mod tests {
|
||||
fn all_themes_distinct_and_labeled() {
|
||||
// ALL covers exactly the variants once, each with a unique non-empty label
|
||||
// and a distinct base colour (so swatches don't look identical).
|
||||
assert_eq!(AppTheme::ALL.len(), 10);
|
||||
assert_eq!(AppTheme::ALL.len(), 13);
|
||||
let mut labels: Vec<&str> = AppTheme::ALL.iter().map(|t| t.label()).collect();
|
||||
labels.sort_unstable();
|
||||
labels.dedup();
|
||||
assert_eq!(labels.len(), 10, "labels must be unique + non-empty");
|
||||
assert_eq!(labels.len(), 13, "labels must be unique + non-empty");
|
||||
assert!(labels.iter().all(|l| !l.is_empty()));
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,354 @@
|
||||
//! Phase-0 spike for post-grace gossip recovery.
|
||||
//!
|
||||
//! These tests prove that `GossipSender::join_peers` can restore an existing
|
||||
//! topic subscription after the other peer drops and rejoins without its own
|
||||
//! bootstrap target. The second case disables relays, clears the surviving
|
||||
//! node's lookup, and moves the peer to a fresh endpoint address so only the
|
||||
//! retained full address passed to `rebootstrap_peers` can drive recovery.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use iroh::address_lookup::memory::MemoryLookup;
|
||||
use iroh::endpoint::presets;
|
||||
use iroh::protocol::Router;
|
||||
use iroh::{Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey};
|
||||
use iroh_gossip::net::Gossip;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
use peerspeak::network::gossip::IrohGossipState;
|
||||
use peerspeak::network::{PeerSpeakTicket, PeerState, RoomEvent, RoomState};
|
||||
|
||||
const EVENT_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
|
||||
struct GossipNode {
|
||||
endpoint: Endpoint,
|
||||
lookup: MemoryLookup,
|
||||
room: Arc<IrohGossipState>,
|
||||
_router: Router,
|
||||
}
|
||||
|
||||
async fn spawn_node(secret: SecretKey) -> GossipNode {
|
||||
let lookup = MemoryLookup::new();
|
||||
let endpoint = Endpoint::builder(presets::Minimal)
|
||||
.secret_key(secret.clone())
|
||||
.relay_mode(RelayMode::Disabled)
|
||||
.address_lookup(lookup.clone())
|
||||
.bind()
|
||||
.await
|
||||
.expect("bind gossip endpoint");
|
||||
let gossip = Gossip::builder().spawn(endpoint.clone());
|
||||
let router = Router::builder(endpoint.clone())
|
||||
.accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone())
|
||||
.spawn();
|
||||
let room = Arc::new(IrohGossipState::new(
|
||||
endpoint.clone(),
|
||||
gossip,
|
||||
lookup.clone(),
|
||||
secret,
|
||||
));
|
||||
|
||||
GossipNode {
|
||||
endpoint,
|
||||
lookup,
|
||||
room,
|
||||
_router: router,
|
||||
}
|
||||
}
|
||||
|
||||
fn state(name: &str, addr: EndpointAddr) -> PeerState {
|
||||
PeerState {
|
||||
name: name.to_string(),
|
||||
is_muted: false,
|
||||
addr,
|
||||
sharing: None,
|
||||
avatar: Default::default(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn await_joined(rx: &mut mpsc::Receiver<RoomEvent>, peer_id: EndpointId) -> PeerState {
|
||||
let deadline = tokio::time::Instant::now() + EVENT_TIMEOUT;
|
||||
loop {
|
||||
match tokio::time::timeout_at(deadline, rx.recv()).await {
|
||||
Ok(Some(RoomEvent::PeerJoined(id, peer_state))) if id == peer_id => return peer_state,
|
||||
Ok(Some(_)) => continue,
|
||||
Ok(None) => panic!("room event channel closed while waiting for PeerJoined"),
|
||||
Err(_) => panic!("timed out waiting for PeerJoined({peer_id:?})"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn await_joined_all(rx: &mut mpsc::Receiver<RoomEvent>, peer_ids: &[EndpointId]) {
|
||||
let deadline = tokio::time::Instant::now() + EVENT_TIMEOUT;
|
||||
let mut remaining = peer_ids.to_vec();
|
||||
while !remaining.is_empty() {
|
||||
match tokio::time::timeout_at(deadline, rx.recv()).await {
|
||||
Ok(Some(RoomEvent::PeerJoined(id, _))) => remaining.retain(|wanted| *wanted != id),
|
||||
Ok(Some(_)) => {}
|
||||
Ok(None) => panic!("room event channel closed while waiting for PeerJoined set"),
|
||||
Err(_) => panic!("timed out waiting for PeerJoined set: {remaining:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn await_left(rx: &mut mpsc::Receiver<RoomEvent>, peer_id: EndpointId) {
|
||||
let deadline = tokio::time::Instant::now() + EVENT_TIMEOUT;
|
||||
loop {
|
||||
match tokio::time::timeout_at(deadline, rx.recv()).await {
|
||||
Ok(Some(RoomEvent::PeerLeft(id))) if id == peer_id => return,
|
||||
Ok(Some(_)) => continue,
|
||||
Ok(None) => panic!("room event channel closed while waiting for PeerLeft"),
|
||||
Err(_) => panic!("timed out waiting for PeerLeft({peer_id:?})"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn await_absent(room: &IrohGossipState, peer_id: EndpointId) {
|
||||
let deadline = tokio::time::Instant::now() + EVENT_TIMEOUT;
|
||||
loop {
|
||||
if !room.active_peers().iter().any(|(id, _)| *id == peer_id) {
|
||||
return;
|
||||
}
|
||||
assert!(
|
||||
tokio::time::Instant::now() < deadline,
|
||||
"timed out waiting for peer to leave the roster"
|
||||
);
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
}
|
||||
}
|
||||
|
||||
fn ticket(host_addr: EndpointAddr) -> String {
|
||||
PeerSpeakTicket {
|
||||
host_addr,
|
||||
topic_id: rand::random(),
|
||||
name: "rebootstrap-spike".to_string(),
|
||||
}
|
||||
.to_string()
|
||||
}
|
||||
|
||||
async fn establish_room(
|
||||
a: &GossipNode,
|
||||
b: &GossipNode,
|
||||
ticket: &str,
|
||||
events_a: &mut mpsc::Receiver<RoomEvent>,
|
||||
) {
|
||||
// B is the ticket host. Its own bootstrap set is empty; A is the only side
|
||||
// that initially dials, which is also how the recovery setup is controlled.
|
||||
b.room
|
||||
.join(ticket, state("Bob", b.endpoint.addr()), vec![])
|
||||
.await
|
||||
.expect("host joins topic");
|
||||
a.room
|
||||
.join(ticket, state("Alice", a.endpoint.addr()), vec![])
|
||||
.await
|
||||
.expect("client joins topic");
|
||||
let joined = await_joined(events_a, b.endpoint.id()).await;
|
||||
assert_eq!(joined.name, "Bob");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rebootstrap_restores_roster_on_existing_subscription() {
|
||||
let a = spawn_node(SecretKey::generate()).await;
|
||||
let b = spawn_node(SecretKey::generate()).await;
|
||||
let ticket = ticket(b.endpoint.addr());
|
||||
let mut events_a = a.room.subscribe_events().await.expect("subscribe A events");
|
||||
|
||||
establish_room(&a, &b, &ticket, &mut events_a).await;
|
||||
|
||||
// Drop only B's topic subscription. A stays subscribed. B then rejoins as
|
||||
// the ticket host, so compute_bootstrap removes self and B has nobody to dial.
|
||||
b.room.leave().await.expect("B leaves topic");
|
||||
await_absent(&a.room, b.endpoint.id()).await;
|
||||
b.room
|
||||
.join(&ticket, state("Bob recovered", b.endpoint.addr()), vec![])
|
||||
.await
|
||||
.expect("B rejoins without bootstrap peers");
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
assert!(
|
||||
!a.room
|
||||
.active_peers()
|
||||
.iter()
|
||||
.any(|(id, _)| *id == b.endpoint.id()),
|
||||
"B must not recover before A explicitly re-bootstraps it"
|
||||
);
|
||||
|
||||
a.room
|
||||
.rebootstrap_peers(vec![b.endpoint.addr()])
|
||||
.await
|
||||
.expect("targeted gossip re-bootstrap");
|
||||
|
||||
let recovered = await_joined(&mut events_a, b.endpoint.id()).await;
|
||||
assert_eq!(recovered.name, "Bob recovered");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rebootstrap_uses_retained_full_address_with_empty_lookup() {
|
||||
let a = spawn_node(SecretKey::generate()).await;
|
||||
let b_secret = SecretKey::generate();
|
||||
let b = spawn_node(b_secret.clone()).await;
|
||||
let b_id = b.endpoint.id();
|
||||
let old_b_addr = b.endpoint.addr();
|
||||
let ticket = ticket(old_b_addr.clone());
|
||||
let mut events_a = a.room.subscribe_events().await.expect("subscribe A events");
|
||||
|
||||
establish_room(&a, &b, &ticket, &mut events_a).await;
|
||||
b.room.leave().await.expect("old B leaves topic");
|
||||
await_absent(&a.room, b_id).await;
|
||||
|
||||
// Move the same authenticated identity to a newly-bound direct-only endpoint.
|
||||
// The old cached path is now dead; the new full address is the only valid one.
|
||||
b.endpoint.close().await;
|
||||
drop(b);
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
let b_rebound = spawn_node(b_secret).await;
|
||||
let new_b_addr = b_rebound.endpoint.addr();
|
||||
assert_eq!(new_b_addr.id, b_id, "identity must survive the rebind");
|
||||
assert_ne!(
|
||||
new_b_addr, old_b_addr,
|
||||
"rebound peer must have a fresh address"
|
||||
);
|
||||
|
||||
b_rebound
|
||||
.room
|
||||
.join(&ticket, state("Bob rebound", new_b_addr.clone()), vec![])
|
||||
.await
|
||||
.expect("rebound host joins without bootstrap peers");
|
||||
|
||||
// Remove the stale lookup entry. `rebootstrap_peers` must seed the retained
|
||||
// new full address before asking gossip to join the peer by id.
|
||||
a.lookup.remove_endpoint_info(b_id);
|
||||
assert!(
|
||||
a.lookup.get_endpoint_info(b_id).is_none(),
|
||||
"A lookup starts empty for B"
|
||||
);
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
assert!(
|
||||
!a.room.active_peers().iter().any(|(id, _)| *id == b_id),
|
||||
"rebound B must not be rediscovered without the retained address"
|
||||
);
|
||||
|
||||
a.room
|
||||
.rebootstrap_peers(vec![new_b_addr])
|
||||
.await
|
||||
.expect("retained-address gossip re-bootstrap");
|
||||
assert!(
|
||||
a.lookup.get_endpoint_info(b_id).is_some(),
|
||||
"re-bootstrap must restore B's address to the lookup"
|
||||
);
|
||||
|
||||
let recovered = await_joined(&mut events_a, b_id).await;
|
||||
assert_eq!(recovered.name, "Bob rebound");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn demoted_peer_requires_a_fresh_signed_announce_to_rejoin() {
|
||||
let a = spawn_node(SecretKey::generate()).await;
|
||||
let b = spawn_node(SecretKey::generate()).await;
|
||||
let ticket = ticket(b.endpoint.addr());
|
||||
let mut events_a = a.room.subscribe_events().await.expect("subscribe A events");
|
||||
|
||||
establish_room(&a, &b, &ticket, &mut events_a).await;
|
||||
a.room.mark_peer_disconnected(b.endpoint.id());
|
||||
assert!(
|
||||
!a.room
|
||||
.active_peers()
|
||||
.iter()
|
||||
.any(|(id, _)| *id == b.endpoint.id()),
|
||||
"demotion must revoke live roster membership"
|
||||
);
|
||||
|
||||
b.room
|
||||
.update_self_state(state("Bob authenticated again", b.endpoint.addr()))
|
||||
.await
|
||||
.expect("broadcast fresh signed announce");
|
||||
|
||||
let recovered = await_joined(&mut events_a, b.endpoint.id()).await;
|
||||
assert_eq!(recovered.name, "Bob authenticated again");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn signed_leave_after_demotion_still_emits_peer_left() {
|
||||
let a = spawn_node(SecretKey::generate()).await;
|
||||
let b = spawn_node(SecretKey::generate()).await;
|
||||
let ticket = ticket(b.endpoint.addr());
|
||||
let mut events_a = a.room.subscribe_events().await.expect("subscribe A events");
|
||||
|
||||
establish_room(&a, &b, &ticket, &mut events_a).await;
|
||||
a.room.mark_peer_disconnected(b.endpoint.id());
|
||||
|
||||
b.room
|
||||
.leave()
|
||||
.await
|
||||
.expect("broadcast signed Leave after demotion");
|
||||
await_left(&mut events_a, b.endpoint.id()).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn targeted_rebootstrap_preserves_healthy_peer_in_three_peer_room() {
|
||||
let a = spawn_node(SecretKey::generate()).await;
|
||||
let b = spawn_node(SecretKey::generate()).await;
|
||||
let c_secret = SecretKey::generate();
|
||||
let c = spawn_node(c_secret.clone()).await;
|
||||
let c_id = c.endpoint.id();
|
||||
let ticket = ticket(c.endpoint.addr());
|
||||
let mut events_a = a.room.subscribe_events().await.expect("subscribe A events");
|
||||
let mut events_b = b.room.subscribe_events().await.expect("subscribe B events");
|
||||
|
||||
c.room
|
||||
.join(&ticket, state("Carol", c.endpoint.addr()), vec![])
|
||||
.await
|
||||
.expect("C hosts topic");
|
||||
b.room
|
||||
.join(&ticket, state("Bob", b.endpoint.addr()), vec![])
|
||||
.await
|
||||
.expect("B joins C");
|
||||
await_joined(&mut events_b, c_id).await;
|
||||
a.room
|
||||
.join(&ticket, state("Alice", a.endpoint.addr()), vec![])
|
||||
.await
|
||||
.expect("A joins C");
|
||||
await_joined_all(&mut events_a, &[b.endpoint.id(), c_id]).await;
|
||||
await_joined(&mut events_b, a.endpoint.id()).await;
|
||||
|
||||
// Remove only C. A and B keep their existing topic subscriptions and remain
|
||||
// mutually present while C is rebound to a fresh address.
|
||||
c.endpoint.close().await;
|
||||
drop(c);
|
||||
a.room.mark_peer_disconnected(c_id);
|
||||
b.room.mark_peer_disconnected(c_id);
|
||||
let c_rebound = spawn_node(c_secret).await;
|
||||
let rebound_addr = c_rebound.endpoint.addr();
|
||||
c_rebound
|
||||
.room
|
||||
.join(
|
||||
&ticket,
|
||||
state("Carol recovered", rebound_addr.clone()),
|
||||
vec![],
|
||||
)
|
||||
.await
|
||||
.expect("rebound C rejoins as host without bootstrap");
|
||||
|
||||
a.room
|
||||
.rebootstrap_peers(vec![rebound_addr])
|
||||
.await
|
||||
.expect("A targets only C for recovery");
|
||||
let recovered = await_joined(&mut events_a, c_id).await;
|
||||
assert_eq!(recovered.name, "Carol recovered");
|
||||
await_joined(&mut events_b, c_id).await;
|
||||
|
||||
assert!(
|
||||
a.room
|
||||
.active_peers()
|
||||
.iter()
|
||||
.any(|(id, _)| *id == b.endpoint.id()),
|
||||
"healthy B must remain present at A throughout C recovery"
|
||||
);
|
||||
assert!(
|
||||
b.room
|
||||
.active_peers()
|
||||
.iter()
|
||||
.any(|(id, _)| *id == a.endpoint.id()),
|
||||
"healthy A must remain present at B throughout C recovery"
|
||||
);
|
||||
}
|
||||
@@ -25,8 +25,7 @@ use peerspeak::codec::opus_impl::OpusEncoder;
|
||||
use peerspeak::core::jitter::{FRAME_SAMPLES, JitterBuffer};
|
||||
use peerspeak::network::{ConnEvent, NetworkTransport};
|
||||
use peerspeak::network::iroh_impl::{AudioRouter, IrohTransport};
|
||||
|
||||
const AUDIO_ALPN: &[u8] = b"peerspeak-audio";
|
||||
use peerspeak::protocol::AUDIO_ALPN;
|
||||
|
||||
struct Node {
|
||||
endpoint: Endpoint,
|
||||
|
||||
Executable
+17
@@ -0,0 +1,17 @@
|
||||
#!/bin/bash
|
||||
# Cross-compile peerspeak to x86_64-pc-windows-gnu (run inside the peerspeak-win distrobox).
|
||||
# Produces a statically-linked, self-contained .exe (no extra DLLs) for the
|
||||
# Windows installer in packaging/windows/. Requires the rust-src component and
|
||||
# the x86_64-pc-windows-gnu target; invoke with build-std for the static link:
|
||||
# RUSTC_BOOTSTRAP=1 ./win-cross-build.sh -Z build-std=std,panic_abort
|
||||
set -euo pipefail
|
||||
cd "$(dirname "$0")"
|
||||
|
||||
export CARGO_TARGET_X86_64_PC_WINDOWS_GNU_LINKER=x86_64-w64-mingw32-gcc
|
||||
export CC_x86_64_pc_windows_gnu=x86_64-w64-mingw32-gcc
|
||||
export CXX_x86_64_pc_windows_gnu=x86_64-w64-mingw32-g++
|
||||
export AR_x86_64_pc_windows_gnu=x86_64-w64-mingw32-ar
|
||||
# Bundled libopus declares cmake_minimum_required < 3.5; cmake 4.x refuses it.
|
||||
export CMAKE_POLICY_VERSION_MINIMUM=3.5
|
||||
|
||||
cargo build --release --target x86_64-pc-windows-gnu --bin peerspeak "$@"
|
||||
Reference in New Issue
Block a user