Docs: Yggdrasil overlay note (rationale, deployment, admin API) + link updates

sn-node: Yggdrasil config generation and admin socket client (genconf, status)
sn-fetch: per-exit egress (Direct vs Proxy) for neighbor exits over the mesh
2026-09-24 16:00:43 +02:00 · 2026-09-24 15:58:12 +02:00 · 2026-09-24 15:58:12 +02:00 · 2026-09-24 15:58:11 +02:00
26 changed files with 2557 additions and 74 deletions
@@ -21,6 +21,7 @@ Nachbarschaft betreibt ihre eigene Instanz nach eigenen Regeln.
 | [[Ethos]] | Werte: Gemeingut, Fairness, Leechen erlaubt, legale Basis |
 | [[Idee]] | Kernidee + ehrlicher Realitätscheck |
 | [[Architektur]] | Schichten: Physik → Routing → Pooling → Freenet-Inhalte |
+| [[Overlay]] | Yggdrasil als Overlay: Auswahl, Deployment, Grenzen |
 | [[Recht]] | Deutsche Rechtslage + Pflichtregeln für die Software |
 | [[Splitting]] | Was sich splitten lässt: Ranges, Flow-Verteilung, Realtime |
 | [[Lizenz]] | Lizenz-Optionen (Code & Doku) |
@@ -41,6 +42,8 @@ Nachbarschaft betreibt ihre eigene Instanz nach eigenen Regeln.
 - [[Recht]] §5 (Pflichtregeln) ↔ [[Server-README]] (Haftungsausschluss)
 - [[Idee]] (Realitätscheck) ↔ [[Architektur]] (link-agnostische Physik, `sn-fetch`)
 - [[Splitting]] ↔ [[Architektur]] (Schicht 2) — was der Exit-Pool tatsächlich nutzen kann
+- [[Overlay]] ↔ [[Architektur]] (Schicht 1) — Routing-Entscheidung und deren Grenzen
+- [[Overlay]] ↔ [[Server-README]] — `sn-node` generiert die Yggdrasil-Config, `sn-fetch` nutzt `Egress::Proxy`
 - [[Lizenz]] ↔ [[Ethos]] §6 (Transparenz) und `freenet-core` (AGPL)

 ## Außerhalb der Doku
@@ -53,7 +53,8 @@ mehr Pooling-Gewinn.

 ## Schicht 1 — Routing (Overlay über alle Haushalte)

-Kandidaten (Auswahl offen, Kriterium: simpel, wartbar, verschlüsselt):
+Auswahl, Deployment und Grenzen stehen in [[Overlay]] — kurz: **Yggdrasil
+ist gewählt** (erste Wahl, bewusst austauschbar gehalten):

 | Kandidat | Pro | Con |
 |---|---|---|
@@ -0,0 +1,134 @@
+# Overlay: Yggdrasil
+
+Wie die Häuser untereinander erreichbar werden, bevor irgendwas gepoolt
+wird. Stand: konzeptionell entschieden (erste Wahl), `sn-node`-Anbindung
+begonnen (siehe [[Server-README]]).
+
+## Warum Yggdrasil?
+
+[Yggdrasil](https://yggdrasilnetwork.github.io/) ist ein
+**self-contained Userspace-Router**: Jeder Node bekommt eine stabilen
+IPv6-Adresse im `200::/7`-Raum, Nodes verbinden sich zu einem
+end-to-end-verschlüsselten Overlay (Routing über einen
+Spanning-Tree), egal ob Glasfaser, LAN, WLAN oder Internet dazwischen
+liegt.
+
+Pro:
+
+- **Out-of-the-box verschlüsselt** — Peering-Verbindungen sind
+  kryptographisch gesichert, ohne extra WireGuard-Stapel
+- **Null Konfiguration fürs Peering**: auf demselben LAN (also über die
+  Nachbarschafts-Faser oder den Keller-Switch) finden sich Nodes per
+  Multicast automatisch
+- **Link-agnostisch** wie SpiderNet selbst: tcp://, tls://, quic://
+  Listener und Peer-URIs, Funk/LAN/Faser egal
+- **Feste Adresse pro Keypair** — die Node-Adresse bleibt, solange der
+  Schlüssel bleibt (wichtig für Exit-Zuteilung und Nachbarschafts-ACL)
+- **`AllowedPublicKeys`**: nur gelistete Nachbarn dürfen peeren —
+  das Nachbarschafts-Vertrauen wird direkt vom Overlay erzwungen
+- Single Rust/CPU-Binary, läuft auf Mini-PCs und Einplatinen
+
+Contra (ehrlich benannt):
+
+- Experimenteller als Babel; Performance-Obergrenze liegt deutlich
+  unter der, die dichte Glasfaser theoretisch bieten könnte — für
+  Phase 1 (2–3 Haushalte) völlig ausreichend, Messwerte dokumentieren
+- Wenn das Overlay zum Flaschenhals wird: Babel + WireGuard als
+  Ausweich-Kandidat (siehe [[Architektur]], Schicht 1)
+
+## Rollen im SpiderNet-Node
+
+```mermaid
+flowchart LR
+    subgraph Haushalt
+        DEVICES[Geraete] -->|own uplink| ROUTER[Router/ISP]
+        NODE[sn-node] --- YG[yggdrasil daemon]
+        YG <-->|mesh links| NACHBAR[yggdrasil der Nachbarn]
+        NODE -->|getSelf / getPeers| YG
+        NODE -->|sn-exit: HTTP proxy| YG
+    end
+    NACHBAR -->| Exit requests| ORIGIN((Origin))
+```
+
+1. **yggdrasil-Daemon** (Prozess/Sidecar, nicht von uns implementiert):
+   bildet das Overlay, jede Faser wird zur Overlay-Verbindung.
+2. **sn-node** (unser Daemon): generiert die yggdrasil.conf aus
+   validierten Einstellungen, fragt den Admin-Socket ab (Node-Adresse,
+   Peers), orchestriert Exit und Fetch.
+3. **sn-exit** (geplant): HTTP-Proxy, der auf der Yggdrasil-Adresse des
+   Haushalts lauscht — Nachbarn senden ihre Segment-Requests dorthin,
+   der Haushalt holt sie über seinen Uplink.
+4. **sn-fetch**: kennt pro Exit einen `Egress` (`Direct` = eigener
+   Uplink, `Proxy` = HTTP-Proxy eines Nachbarn) und plant Segmente
+   darauf.
+
+## Admin-Socket (technische Fakten)
+
+Quelle: [Admin-API-Doku](https://yggdrasil-network.github.io/admin.html)
+
+- Default `localhost:9001` (TCP), alternativ Unix-Socket
+  (`AdminListen` in der Konfig)
+- Protokoll: JSON-Stanza + `\n`; Anfrage `{"request": "getself"}`;
+  Antwort enthält immer `"status"` (`"success"`/`"error"`), optional
+  `"response"` und `"error"`
+- `getSelf` → eigene Adresse, routed `/64`-Subnetz (`300:…::/64`),
+  Public Key, Build-Version
+- `getPeers` → aktive Peer-Sessions inkl. Peering-URI
+- `sn-node status` kapselt genau das (siehe unten)
+
+## Deployment: zwei Häuser verkabeln
+
+Schritt-für-Schritt für Phase 1 (Prototyp-Scope, **keine Garantie**,
+siehe [[Recht]]):
+
+1. **Hardware:** je Haus ein Mini-PC mit zwei NICs (eine zum
+   Heim-Router, eine zur Nachbarschafts-Faser bzw. zum Switch im
+   Keller) — oder übergangsweise eine NIC am Mesh-Switch.
+2. **Yggdrasil installieren** (Debian-Paket oder Binary).
+3. **Config generieren:**
+   `sn-node genconf --peer "tcp://[<Nachbar-Adresse>]:1337" --listen "tcp://[::]:1337" --allow-key <Nachbar-Public-Key> --out /etc/yggdrasil.conf`
+   - im selben LAN reicht Multicast-Auto-Peering ohne `--peer`
+   - `--no-multicast`, wenn Peering nur über die explizite Liste laufen
+     soll (z. B. gemischte Nachbarschaft)
+   - die `allowed_public_keys`-ACL: nur die gelisteten Nachbarn können
+     peeren — für ein Viertel die persönliche Freischaltung
+4. **Start:** `yggdrasil -useconffile /etc/yggdrasil.conf`
+5. **Status:** `sn-node status` → eigene Overlay-Adresse, Subnetz,
+   Version, verbundene Nachbarn
+6. **Test:** von Haus A `ping <Adresse von Haus B>` (IPv6 im Overlay).
+7. **Exit anbieten** (geplanter `sn-exit`): HTTP-Proxy auf der
+   Overlay-Adresse binden; beim Nachbarn dann
+   `sn-fetch <url> -e "1=1,2=1@http://<Nachbar-Adresse>:8080"`.
+
+Die Config-Datei enthält den **privaten** Schlüssel des Nodes:
+`chmod 600` (macht `sn-node genconf` selbst), niemals committen, nie
+teilen. Private Adressen realer Nachbarn gehören nicht ins Repo
+(Regeln in `AGENTS.md`).
+
+## Grenzen & Sicherheit
+
+- **Yggdrasil ist keine Anonymisierung.** Die Overlay-Adresse ist ein
+  stabiler Pseudonym; wer Peering-Verbindungen beobachtet, sieht, wer
+  mit wem verbunden ist. SpiderNet verspricht deshalb nirgends
+  Anonymität (konsistent mit Freenet-Position in [[Ethos]]).
+- **Exit = Vertrauensstellung:** Der Exit-Haushalt sieht die URLs der
+  Segmente, die er für Nachbarn holt. Standard-Kontingente, jederzeit
+  widerruflich (Kill-Switch, siehe [[Architektur]]).
+- **Port-Forwarding:** Eingehende Peerings über das Internet (z. B.
+  wenn zwei Häuser nicht direkt verbunden sind und den Weg übers
+  normale Internet nehmen) brauchen ein Forward des `Listen`-Ports;
+  innerhalb der Faser/Multicast nicht nötig.
+- **Wenn die Faser später schneller als Yggdrasil wird:** Messwerte
+  sammeln; ggf. pro Link WireGuard-Punkt-zu-Punkt direkt nutzen und
+  Yggdrasil nur als Control-Plane — Architektur-Schicht 1 ist bewusst
+  austauschbar gehalten.
+
+## Quellen
+
+- Yggdrasil-Website & Docs: <https://yggdrasilnetwork.github.io/>
+- Admin API: <https://yggdrasil-network.github.io/admin.html>
+- Configuration: <https://yggdrasil-network.github.io/configuration.html>
+  (genconf/useconf, HJSON/JSON, `Peers`, `Listen`, `MulticastInterfaces`,
+  `AllowedPublicKeys`, `IfName: none` für router-only)
+- Quellcode (Go, AGPL-3.0 — wir rufen nur das Binary auf, keine
+  Derivate): <https://github.com/yggdrasil-network/yggdrasil-go>
@@ -18,8 +18,14 @@ Ziel: Beweis, dass Pooling über zwei Anschlüsse real funktioniert.
 - [x] `sn-fetch` PoC (lokal): Range-Probe, gewichteter Planner, parallele
      Segment-Downloads, Reassembly + SHA-256 — verifiziert gegen einen
      echten Server (Debian-Mirror); Multi-Exit-Routing folgt mit dem Mesh
- [ ] SpiderNet-Node auf Mini-PC (Raspberry Pi / N100-Box), Yggdrasil-Overlay
-      zwischen 2 Haushalten (Kabel oder Kurzstrecke WLAN/60-GHz)
+- [x] `sn-fetch`: Segment-Retry mit Exit-Reassignment, Progress-Sinks,
+      per-Exit-Egress (`Direct`/`Proxy`) — Nachbar-Exit über HTTP-Proxy
+      end-to-end getestet
+- [x] `sn-node` v0: Yggdrasil-Config-Generierung (`genconf`) + Admin-Socket-
+      Client (`status`) — die Overlay-Anbindung beginnt
+- [ ] Yggdrasil-Overlay zwischen 2 Haushalten (Kabel oder Kurzstrecke
+      WLAN/60-GHz), `sn-node`-Daemon orchestriert den Node
+- [ ] `sn-exit` v0: HTTP-Proxy auf der Yggdrasil-Adresse (Kontingente)
 - [ ] `sn-fetch` PoC über 2 echte Exits, plus Baseline-Messung
      (1-Exit vs. 2-Exit, gleiche Datei)
 - [ ] Fair-Share-Scheduler v0: zwei gleichzeitige Downloads, faire Aufteilung
@@ -29,8 +35,8 @@ Ziel: Beweis, dass Pooling über zwei Anschlüsse real funktioniert.

 Ziel: ein Paket, das 3–10 Haushalte installieren können.

- [ ] `sn-node` Daemon: Discovery, Peering, Metriken (per-Exit Bandbreite/
-      Latenz), Onboarding-Flow (QR/Knopfdruck)
+- [ ] `sn-node` Daemon v1: Metriken aus Yggdrasil-Peering (per-Exit
+      Bandbreite/Latenz), Onboarding-Flow (QR/Knopfdruck)
 - [ ] Exit-Pool: Kontingente, opt-in/opt-out, Leech-Modus, Kill-Switch
 - [ ] `sn-fetch`: API + CLI + einfache Browser-Integration
 - [ ] `sn-cache`: lancache-artiger transparenter Cache (DNS-basiert)
@@ -1,6 +1,6 @@
 [workspace]
 resolver = "2"
-members = ["crates/sn-fetch"]
+members = ["crates/sn-fetch", "crates/sn-node"]

 [workspace.package]
 version = "0.1.0"
@@ -18,7 +18,8 @@ panic = "deny"
 unreachable = "deny"

 [workspace.dependencies]
-tokio = { version = "1", features = ["rt-multi-thread", "macros", "fs", "io-util"] }
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "fs", "io-util", "net", "time"] }
+serde_json = "1"
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }
 sha2 = "0.10"
 hex = "0.4"
@@ -12,17 +12,39 @@ macht. Sprache: **Rust**.
  Exit, Mindest-Segmentgröße mit Merge-Fix, validated: lückenlose,
  vollständige Abdeckung)
 - Parallele Segment-Downloads (Streaming, eigener reqwest-Client pro Exit)
+- **Retry mit Exit-Reassignment:** fehlgeschlagene Segmente werden (wieder-
+  holbar, Standard 3 Versuche) bevorzugt auf einem anderen Exit wiederholt;
+  nicht-retrybare Fehler (4xx, korrupte Parts) brechen sofort ab — ein
+  halbfertiges File wird nie als Erfolg ausgegeben
+- **Progress:** `ProgressSink`-Events (Segment started/finished/failed,
+  `BytesTracker`) + `tracing`-Logging
+- **Per-Exit-Egress:** `Direct` (eigener Uplink) oder `Proxy` (HTTP-Proxy
+  eines Nachbarn — der Weg für `sn-exit` über das Overlay); end-to-end
+  getestet gegen einen echten Forward-Proxy
 - Reassembly in Index-Reihenfolge + Streaming-SHA-256 + Längen-Check
 - Fallback: range-hostile Server werden als single-stream geladen
- CLI: `sn-fetch <url> [-o out] [-e "1=2,2=1"] [-s n] [--sha256 hex] --probe-only`
+- CLI: `sn-fetch <url> [-o out] [-e "1=2,2=1@http://[ygg]:8080"] [-s n]
+  [--max-attempts n] [--sha256 hex] --probe-only`

 Verifiziert gegen einen echten Server (Debian-CD-Mirror, 756 MB, 8 Segmente,
 SHA-256 entspricht der offiziellen `SHA256SUMS`).

-**Noch nicht drin:** Multi-Exit-Routing über das Mesh (aktuell laufen alle
-Exits als lokale Verbindungen — die Struktur [`HttpExit`]-Trait,
-Exit-Registry] ist dafür schon bereit), Retry bei Segment-Fehlern,
-Progress-Anzeige während des Laufs.
+**`sn-node` v0 ist implementiert** (siehe `crates/sn-node/`):
+
+- `sn-node genconf`: erzeugt eine Yggdrasil-Config aus validierten
+  Einstellungen — Basis ist immer der echte `yggdrasil -genconf -json`
+  Output (keine hartcodierten Defaults), gepatcht mit `Peers`, `Listen`,
+  `MulticastInterfaces`, `AdminListen`, `AllowedPublicKeys` (Nachbarschafts-
+  ACL), `IfName`, `PrivateKeyPath`; Output-Datei wird auf 600 gesetzt
+- `sn-node status`: fragt den Admin-Socket ab (`getSelf`/`getPeers` über
+  TCP `localhost:9001` oder Unix-Socket) — Overlay-Adresse, Subnetz,
+  Version, aktive Peer-Verbindungen
+
+**Noch nicht drin:** `sn-exit` (der Proxy-Dienst, der auf der Yggdrasil-
+Adresse lauscht und Segmente für Nachbarn holt), `sn-node`-Daemon (Supervisor,
+der yggdrasil + sn-exit + sn-fetch orchestriert), Fairness-Scheduler.
+
+Details zum Overlay: [[Overlay]].

 ## Haftungsausschluss (muss im Produkt sichtbar sein)

@@ -52,14 +74,14 @@ Einplatinenrechner), der:

 | Crate/Binary | Zweck |
 |---|---|
-| `sn-node` | Daemon: Discovery, Peering, Metriken, Onboarding |
-| `sn-exit` | Exit-Dienst: kontingentierter Uplink-Proxy für Nachbarn (verschlüsselt) |
+| `sn-node` | Node-Glue: Yggdrasil-Config (`genconf`), Overlay-Status (`status`), später Daemon/Supervisor |
+| `sn-exit` | Exit-Dienst: kontingentierter Uplink-Proxy auf der Yggdrasil-Adresse (geplant) |
 | `sn-fetch` | Segmentierter Multi-Exit-Downloader (HTTP-API + CLI) |
 | `sn-fair` | Fairness-Scheduler (max-min, Exit-Kontingente, Karma, Leech-Modus) |
 | `sn-cache` | Transparenter Nachbarschafts-Cache (DNS-basiert) |
 | `sn-freenet` | Integration von `freenet-core` (Contracts für Nachbarschafts-Inhalte) |

-## MVP (Phase 2 der [[Roadmap]])
+## MVP (Phase 1→2, siehe [[Roadmap]])

 Kleinster brauchbarer Schnitt:

@@ -47,4 +47,12 @@ pub enum FetchError {

    #[error("invalid sha256 hex string")]
    InvalidDigest,
+
+    #[error("segment {index}: all {attempts} attempt(s) failed; last error: {last_error}")]
+    SegmentAttemptsExhausted {
+        index: SegmentIndex,
+        attempts: u32,
+        #[source]
+        last_error: Box<FetchError>,
+    },
 }
@@ -1,7 +1,18 @@
-//! Execution: download planned segments in parallel and reassemble the
-//! target file.
+//! Execution: download planned segments in parallel (with per-segment
+//! retry and exit reassignment) and reassemble the target file.
+//!
+//! API note: [`SegmentStats`] keeps its original shape and `run_segments`
+//! keeps its signature, so existing callers (runner.rs constructs
+//! `SegmentStats` literals directly) stay source-compatible. The
+//! retry-aware path [`run_segments_with_policy`] returns
+//! [`SegmentAttemptStats`] instead, which additionally records the exit
+//! that served the successful attempt and how many attempts (retries
+//! included) the segment needed; `run_segments` converts via `From`.

+use std::collections::BTreeMap;
+use std::num::NonZeroU32;
 use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex, MutexGuard};
 use std::time::{Duration, Instant};

 use sha2::{Digest, Sha256};
@@ -10,6 +21,7 @@ use tokio::io::AsyncWriteExt;
 use crate::error::FetchError;
 use crate::exit::ExitRegistry;
 use crate::plan::FetchPlan;
+use crate::progress::{ProgressEvent, ProgressSink};
 use crate::types::{ByteRange, ExitId, FileSize, SegmentIndex, Sha256Digest, TargetUrl};

 /// Statistics for one finished segment.
@@ -20,6 +32,50 @@ pub struct SegmentStats {
    pub duration: Duration,
 }

+/// Retry behaviour for segment downloads.
+#[derive(Debug, Clone, Copy)]
+pub struct RetryPolicy {
+    /// Total tries per segment; attempt 1 is the initial try.
+    pub max_attempts: NonZeroU32,
+}
+
+impl RetryPolicy {
+    #[must_use]
+    pub const fn new(max_attempts: NonZeroU32) -> Self {
+        Self { max_attempts }
+    }
+}
+
+impl Default for RetryPolicy {
+    fn default() -> Self {
+        Self {
+            max_attempts: NonZeroU32::new(3).unwrap_or(NonZeroU32::MIN),
+        }
+    }
+}
+
+/// Statistics for one finished segment, including retry information.
+/// `duration` spans all attempts (retries included); `exit` is the exit
+/// that served the successful attempt.
+#[derive(Debug, Clone)]
+pub struct SegmentAttemptStats {
+    pub index: SegmentIndex,
+    pub exit: ExitId,
+    pub bytes: u64,
+    pub duration: Duration,
+    pub attempts: u32,
+}
+
+impl From<SegmentAttemptStats> for SegmentStats {
+    fn from(stats: SegmentAttemptStats) -> Self {
+        SegmentStats {
+            index: stats.index,
+            bytes: stats.bytes,
+            duration: stats.duration,
+        }
+    }
+}
+
 /// Result of a complete fetch.
 #[derive(Debug, Clone)]
 pub struct FetchReport {
@@ -41,19 +97,63 @@ pub async fn run_segments(
    exits: &ExitRegistry,
    parts_dir: &Path,
 ) -> Result<Vec<SegmentStats>, FetchError> {
+    let stats = run_segments_with_policy(
+        plan,
+        exits,
+        parts_dir,
+        &RetryPolicy::default(),
+        &ProgressSink::noop(),
+    )
+    .await?;
+    Ok(stats.into_iter().map(SegmentStats::from).collect())
+}
+
+/// Download every planned segment in parallel, retrying failed segments
+/// according to `policy` — preferably on a different exit — and report
+/// progress to `progress`. Each segment is written to its own part file
+/// inside `parts_dir` (named `seg-NNNNNN.part`); the part file of a
+/// failed attempt is removed before the next attempt.
+///
+/// Failures abort the whole fetch: a segment that exhausts all attempts
+/// fails the fetch with [`FetchError::SegmentAttemptsExhausted`], a
+/// non-retryable failure aborts immediately. A partially assembled file
+/// is never presented as success.
+pub async fn run_segments_with_policy(
+    plan: &FetchPlan,
+    exits: &ExitRegistry,
+    parts_dir: &Path,
+    policy: &RetryPolicy,
+    progress: &ProgressSink,
+) -> Result<Vec<SegmentAttemptStats>, FetchError> {
    tokio::fs::create_dir_all(parts_dir).await?;

+    // Validate that every planned exit exists, then snapshot the
+    // clients of all registered exits: tasks must fetch clients by id
+    // without borrowing the registry ('static spawn requirement).
+    for seg in &plan.segments {
+        exits.get(seg.exit)?;
+    }
+    let mut clients: BTreeMap<ExitId, reqwest::Client> = BTreeMap::new();
+    for id in exits.ids() {
+        clients.insert(id, exits.get(id)?.client().clone());
+    }
+    let clients = Arc::new(clients);
+    let selector = Arc::new(ExitSelector::new(exits));
+
    let mut handles = Vec::with_capacity(plan.segments.len());
    for seg in &plan.segments {
-        let client = exits.get(seg.exit)?.client().clone();
-        handles.push(tokio::spawn(download_segment(
-            plan.url.clone(),
-            seg.exit,
-            seg.index,
-            seg.range,
-            client,
-            parts_dir.to_path_buf(),
-        )));
+        let task = SegmentTask {
+            url: plan.url.clone(),
+            index: seg.index,
+            range: seg.range,
+            planned_exit: seg.exit,
+            clients: Arc::clone(&clients),
+            selector: Arc::clone(&selector),
+            parts_dir: parts_dir.to_path_buf(),
+            policy: *policy,
+            progress: progress.clone(),
+        };
+        handles.push(tokio::spawn(download_with_retry(task)));
    }

    let mut stats = Vec::with_capacity(handles.len());
@@ -67,18 +167,158 @@ pub async fn run_segments(
    Ok(stats)
 }

-/// Download one byte range and write it to `parts_dir/seg-NNNNNN.part`.
-async fn download_segment(
+/// Everything one segment task needs, bundled to keep the task
+/// signature small.
+struct SegmentTask {
    url: TargetUrl,
+    index: SegmentIndex,
+    range: ByteRange,
+    planned_exit: ExitId,
+    clients: Arc<BTreeMap<ExitId, reqwest::Client>>,
+    selector: Arc<ExitSelector>,
+    parts_dir: PathBuf,
+    policy: RetryPolicy,
+    progress: ProgressSink,
+}
+
+/// Download one segment with up to `policy.max_attempts` attempts,
+/// reassigning to a different exit after a retryable failure (while the
+/// registry offers more than one exit).
+async fn download_with_retry(task: SegmentTask) -> Result<SegmentAttemptStats, FetchError> {
+    let SegmentTask {
+        url,
+        index,
+        range,
+        planned_exit,
+        clients,
+        selector,
+        parts_dir,
+        policy,
+        progress,
+    } = task;
+    let max_attempts = policy.max_attempts.get();
+    let started = Instant::now();
+    let mut exit = planned_exit;
+    let mut attempt: u32 = 0;
+
+    loop {
+        attempt += 1;
+        let client = clients
+            .get(&exit)
+            .ok_or(FetchError::UnknownExit(exit))?
+            .clone();
+        progress.on_event(&ProgressEvent::SegmentStarted {
+            index,
+            exit,
+            attempt,
+        });
+        tracing::debug!(segment = index.value(), exit = %exit, attempt, "segment attempt started");
+
+        match download_attempt(&url, exit, index, range, client, &parts_dir).await {
+            Ok(bytes) => {
+                let duration = started.elapsed();
+                progress.on_event(&ProgressEvent::SegmentFinished {
+                    index,
+                    exit,
+                    bytes,
+                    duration,
+                });
+                tracing::info!(
+                    segment = index.value(),
+                    exit = %exit,
+                    attempts = attempt,
+                    bytes,
+                    "segment finished"
+                );
+                return Ok(SegmentAttemptStats {
+                    index,
+                    exit,
+                    bytes,
+                    duration,
+                    attempts: attempt,
+                });
+            }
+            Err(err) => {
+                let retryable = is_retryable(&err);
+                let err_text = err.to_string();
+                progress.on_event(&ProgressEvent::SegmentFailed {
+                    index,
+                    exit,
+                    attempt,
+                    error: err_text,
+                });
+                remove_part_file(&parts_dir, index).await;
+
+                if !retryable {
+                    tracing::warn!(
+                        segment = index.value(),
+                        exit = %exit,
+                        attempt,
+                        "segment failed with non-retryable error: {err}"
+                    );
+                    return Err(err);
+                }
+                selector.mark_failed(exit);
+                if attempt >= max_attempts {
+                    tracing::warn!(
+                        segment = index.value(),
+                        attempts = attempt,
+                        "segment failed after all attempts were exhausted: {err}"
+                    );
+                    return Err(FetchError::SegmentAttemptsExhausted {
+                        index,
+                        attempts: attempt,
+                        last_error: Box::new(err),
+                    });
+                }
+                tracing::warn!(
+                    segment = index.value(),
+                    exit = %exit,
+                    attempt,
+                    "segment attempt failed; retrying: {err}"
+                );
+                exit = selector.pick_retry(exit);
+            }
+        }
+    }
+}
+
+/// Retryable: transport-level failures (network, timeouts), 5xx
+/// responses and length mismatches. Non-retryable: 4xx range rejections
+/// (e.g. 416), corrupt parts, plan or digest errors — retrying those on
+/// another exit cannot succeed.
+fn is_retryable(error: &FetchError) -> bool {
+    match error {
+        FetchError::Http(err) => {
+            !err.is_status() || err.status().is_some_and(|status| status.is_server_error())
+        }
+        FetchError::Io(_) | FetchError::SegmentLengthMismatch { .. } => true,
+        FetchError::RangeRejected { status, .. } => *status >= 500,
+        FetchError::InvalidUrl(_)
+        | FetchError::ZeroFileSize
+        | FetchError::TotalLengthMismatch { .. }
+        | FetchError::HashMismatch { .. }
+        | FetchError::InvalidPlan(_)
+        | FetchError::UnknownExit(_)
+        | FetchError::CorruptPart(_)
+        | FetchError::InvalidDigest
+        | FetchError::SegmentAttemptsExhausted { .. } => false,
+    }
+}
+
+/// Download one byte range and write it to `parts_dir/seg-NNNNNN.part`.
+/// Returns the number of bytes written. A failed attempt can leave a
+/// partial part file behind; the retry loop removes it before the next
+/// attempt.
+async fn download_attempt(
+    url: &TargetUrl,
    exit: ExitId,
    index: SegmentIndex,
    range: ByteRange,
    client: reqwest::Client,
-    parts_dir: PathBuf,
-) -> Result<SegmentStats, FetchError> {
-    let started = Instant::now();
-
-    let part_path = part_file(&parts_dir, index);
+    parts_dir: &Path,
+) -> Result<u64, FetchError> {
+    let part_path = part_file(parts_dir, index);
    let mut response = client
        .get(url.inner().clone())
        .header(reqwest::header::RANGE, range.http_value())
@@ -106,7 +346,7 @@ async fn download_segment(
    file.flush().await?;

    if written != range.len() {
-        tokio::fs::remove_file(&part_path).await.ok();
+        let _ = tokio::fs::remove_file(&part_path).await;
        return Err(FetchError::SegmentLengthMismatch {
            index,
            got: written,
@@ -114,11 +354,68 @@ async fn download_segment(
        });
    }

-    Ok(SegmentStats {
-        index,
-        bytes: written,
-        duration: started.elapsed(),
-    })
+    Ok(written)
+}
+
+/// Chooses exits for retry attempts across all segment tasks of one
+/// run. Preference: the exit that failed least recently (exits that
+/// never failed first, ties broken by ascending id). The exit an
+/// attempt just failed on is avoided while at least one other exit
+/// exists; with a single-exit registry, retries stay on that exit.
+#[derive(Debug)]
+struct ExitSelector {
+    ids: Vec<ExitId>,
+    last_failed: Mutex<BTreeMap<ExitId, Instant>>,
+}
+
+impl ExitSelector {
+    fn new(exits: &ExitRegistry) -> Self {
+        Self {
+            ids: exits.ids(),
+            last_failed: Mutex::new(BTreeMap::new()),
+        }
+    }
+
+    fn mark_failed(&self, id: ExitId) {
+        let mut guard = self.lock();
+        guard.insert(id, Instant::now());
+    }
+
+    fn pick_retry(&self, avoid: ExitId) -> ExitId {
+        let guard = self.lock();
+        let mut best: Option<ExitId> = None;
+        for &id in &self.ids {
+            if id == avoid {
+                continue;
+            }
+            best = Some(match best {
+                None => id,
+                Some(current) => {
+                    let prefer_id = match (guard.get(&current), guard.get(&id)) {
+                        // tie or current never failed: keep the lower id
+                        (None, None | Some(_)) => false,
+                        (Some(_), None) => true, // id never failed: better
+                        (Some(current_failed), Some(id_failed)) => *id_failed < *current_failed,
+                    };
+                    if prefer_id {
+                        id
+                    } else {
+                        current
+                    }
+                }
+            });
+        }
+        best.unwrap_or(avoid) // single-exit registry: retry on the same exit
+    }
+
+    /// std mutexes can be poisoned by a panicking holder; since the
+    /// map is only advisory (exit preference), keep using its contents.
+    fn lock(&self) -> MutexGuard<'_, BTreeMap<ExitId, Instant>> {
+        match self.last_failed.lock() {
+            Ok(guard) => guard,
+            Err(poisoned) => poisoned.into_inner(),
+        }
+    }
 }

 /// Reassemble part files (in index order) into the output file while
@@ -199,6 +496,11 @@ async fn collect_part_files(parts_dir: &Path) -> Result<Vec<PathBuf>, FetchError
    Ok(out)
 }

+/// Best-effort removal of a failed attempt's part file.
+async fn remove_part_file(parts_dir: &Path, index: SegmentIndex) {
+    let _ = tokio::fs::remove_file(part_file(parts_dir, index)).await;
+}
+
 fn part_file(parts_dir: &Path, index: SegmentIndex) -> PathBuf {
    parts_dir.join(format!("seg-{:06}.part", index.value()))
 }
@@ -48,6 +48,37 @@ impl HttpExit for LocalExit {
    }
 }

+/// A neighbor's uplink, reached through an HTTP proxy over the mesh.
+/// The proxy is expected to listen on the neighbor's Yggdrasil address,
+/// so the transport between the households is encrypted end-to-end by
+/// the overlay.
+#[derive(Debug)]
+pub struct MeshExit {
+    id: ExitId,
+    client: reqwest::Client,
+}
+
+impl MeshExit {
+    pub fn new(id: ExitId, proxy: reqwest::Url) -> Result<Self, FetchError> {
+        let proxy = reqwest::Proxy::all(proxy)?;
+        let client = reqwest::Client::builder()
+            .connect_timeout(Duration::from_secs(30))
+            .proxy(proxy)
+            .build()?;
+        Ok(Self { id, client })
+    }
+}
+
+impl HttpExit for MeshExit {
+    fn id(&self) -> ExitId {
+        self.id
+    }
+
+    fn client(&self) -> &reqwest::Client {
+        &self.client
+    }
+}
+
 /// All exits available for a fetch, keyed by id.
 #[derive(Default, Clone)]
 pub struct ExitRegistry {
@@ -18,5 +18,6 @@ pub mod exec;
 pub mod exit;
 pub mod plan;
 pub mod probe;
+pub mod progress;
 pub mod runner;
 pub mod types;
@@ -15,7 +15,7 @@ use std::path::PathBuf;
 use clap::Parser;
 use sn_fetch::runner::{self, FetchOptions};
 use sn_fetch::types::{
-    ExitId, ExitWeight, MinSegmentBytes, SegmentsPerExit, Sha256Digest, TargetUrl,
+    Egress, ExitId, ExitWeight, MinSegmentBytes, SegmentsPerExit, Sha256Digest, TargetUrl,
 };

 #[derive(Debug, Parser)]
@@ -28,12 +28,20 @@ struct Args {
    #[arg(short, long)]
    output: Option<PathBuf>,

-    /// Exits as `id=weight` pairs, comma separated, e.g. `local=1,local2=2`.
-    /// `PoC`: all ids map to plain local connections; the weights only shape
-    /// the plan. Multi-exit routing comes with the mesh integration.
-    #[arg(short = 'e', long = "exit", default_value = "local=1")]
+    /// Exits as `id=weight` pairs, comma separated, e.g. `1=2,2=1`.
+    /// A proxy route turns an exit into a neighbor's uplink over the
+    /// mesh overlay: `id=weight@http://[ygg-addr]:8080`. Without a
+    /// proxy, the exit is this machine's own connection.
+    /// `PoC`: weights only shape the plan; a real proxy service
+    /// (`sn-exit`) comes with the mesh integration.
+    #[arg(short = 'e', long = "exit", default_value = "1=1")]
    exits: String,

+    /// Total tries per segment (initial try + retries; failed segments
+    /// are retried on a different exit when available).
+    #[arg(long, default_value_t = 3)]
+    max_attempts: u32,
+
    /// Parallel sub-segments per exit.
    #[arg(short = 's', long, default_value_t = 1)]
    segments_per_exit: usize,
@@ -51,10 +59,10 @@ struct Args {
    probe_only: bool,
 }

-fn parse_exits(spec: &str) -> Result<Vec<(ExitId, ExitWeight)>, String> {
+fn parse_exits(spec: &str) -> Result<Vec<(ExitId, ExitWeight, Egress)>, String> {
    spec.split(',')
        .map(|pair| {
-            let (id, weight) = pair
+            let (id, rest) = pair
                .split_once('=')
                .ok_or_else(|| format!("invalid exit spec `{pair}`, expected `id=weight`"))?;
            let id = id
@@ -63,13 +71,21 @@ fn parse_exits(spec: &str) -> Result<Vec<(ExitId, ExitWeight)>, String> {
                .ok()
                .and_then(NonZeroU8::new)
                .ok_or_else(|| format!("invalid exit id `{id}` (must be a non-zero number)"))?;
+            let (weight, egress) = match rest.split_once('@') {
+                Some((weight, proxy)) => {
+                    let egress = Egress::proxy(proxy.trim())
+                        .map_err(|err| format!("invalid proxy in exit spec `{pair}`: {err}"))?;
+                    (weight, egress)
+                }
+                None => (rest, Egress::Direct),
+            };
            let weight = weight
                .trim()
                .parse::<u32>()
                .ok()
                .and_then(NonZeroU32::new)
                .ok_or_else(|| format!("invalid exit weight `{weight}` (must be non-zero)"))?;
-            Ok((ExitId::new(id), ExitWeight::new(weight)))
+            Ok((ExitId::new(id), ExitWeight::new(weight), egress))
        })
        .collect()
 }
@@ -131,6 +147,8 @@ async fn run() -> Result<(), String> {
        NonZeroU64::new(args.min_segment_mib.saturating_mul(1024 * 1024))
            .ok_or("min_segment_mib must be at least 1")?,
    );
+    let max_attempts =
+        NonZeroU32::new(args.max_attempts).ok_or("max_attempts must be at least 1")?;

    let expected = match &args.sha256 {
        Some(hex) => Some(Sha256Digest::from_hex(hex).map_err(|err| err.to_string())?),
@@ -146,9 +164,10 @@ async fn run() -> Result<(), String> {
        segments_per_exit: SegmentsPerExit::new(spe),
        min_segment,
        expected_sha256: expected,
+        retries: sn_fetch::exec::RetryPolicy::new(max_attempts),
    };

-    let registry = runner::local_registry(&exit_specs).map_err(|err| err.to_string())?;
+    let registry = runner::registry_from_specs(&exit_specs).map_err(|err| err.to_string())?;
    let parts_dir = std::env::temp_dir().join("sn-fetch-parts");

    let report = runner::run(&target, &options, &registry, &parts_dir)
@@ -2,8 +2,8 @@

 use crate::error::FetchError;
 use crate::types::{
-    ByteRange, ExitId, ExitWeight, FileSize, MinSegmentBytes, SegmentIndex, SegmentsPerExit,
-    TargetUrl,
+    ByteRange, Egress, ExitId, ExitWeight, FileSize, MinSegmentBytes, SegmentIndex,
+    SegmentsPerExit, TargetUrl,
 };
 use std::num::NonZeroUsize;

@@ -23,8 +23,9 @@ pub struct FetchPlan {
    pub segments: Vec<SegmentPlan>,
 }

-/// Exits with their capacity weights.
-pub type ExitSpecs = Vec<(ExitId, ExitWeight)>;
+/// exit description, including how traffic leaves (`Direct` = own
+/// uplink, `Proxy` = a neighbor's exit over the mesh).
+pub type ExitSpecs = Vec<(ExitId, ExitWeight, Egress)>;

 impl FetchPlan {
    /// Build a validated plan.
@@ -56,12 +57,12 @@ impl FetchPlan {
            let weakest = exits
                .iter()
                .enumerate()
-                .min_by_key(|(_, (id, weight))| (weight.value(), std::cmp::Reverse(id.value())))
+                .min_by_key(|(_, (id, weight, _))| (weight.value(), std::cmp::Reverse(id.value())))
                .map_or(0, |(pos, _)| pos);
            exits.remove(weakest);
        }
        // Deterministic boundary assignment order.
-        exits.sort_by_key(|(id, _)| *id);
+        exits.sort_by_key(|(id, _, _)| *id);

        // If even one segment per exit would be too small, reduce to one
        // segment per exit.
@@ -85,7 +86,7 @@ impl FetchPlan {
            // Weighted split across exits, then even sub-split per exit.
            let units: Vec<u128> = exits
                .iter()
-                .map(|(_, w)| u128::from(w.value()) * spe.value() as u128)
+                .map(|(_, w, _)| u128::from(w.value()) * spe.value() as u128)
                .collect();
            let total_units: u128 = units.iter().sum();

@@ -0,0 +1,171 @@
+//! Progress reporting for segmented downloads.
+//!
+//! [`ProgressSink`] is a cloneable callback handle rather than a trait:
+//! spawned segment tasks need an owned, `'static` reporter, and a
+//! callback handle can be cloned cheaply from a plain `&ProgressSink`
+//! parameter without extra plumbing. [`ProgressEvent`] carries one
+//! event per segment attempt; [`BytesTracker`] aggregates the bytes
+//! finished across all segments.
+
+#![allow(clippy::module_name_repetitions)] // `Progress*` types read naturally inside `mod progress`
+
+use std::fmt;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use crate::types::{ExitId, SegmentIndex};
+
+/// One event describing the state of a segment download attempt.
+#[derive(Debug, Clone)]
+pub enum ProgressEvent {
+    /// A segment attempt started (attempt 1 is the initial try).
+    SegmentStarted {
+        index: SegmentIndex,
+        exit: ExitId,
+        attempt: u32,
+    },
+    /// A segment finished successfully. `duration` spans the whole
+    /// segment, all attempts (retries included).
+    SegmentFinished {
+        index: SegmentIndex,
+        exit: ExitId,
+        bytes: u64,
+        duration: Duration,
+    },
+    /// A segment attempt failed. `error` is the formatted failure
+    /// reason; the typed error surfaces through the fetch's result.
+    SegmentFailed {
+        index: SegmentIndex,
+        exit: ExitId,
+        attempt: u32,
+        error: String,
+    },
+}
+
+impl fmt::Display for ProgressEvent {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::SegmentStarted {
+                index,
+                exit,
+                attempt,
+            } => {
+                write!(f, "segment {index} on {exit}: attempt {attempt} started")
+            }
+            Self::SegmentFinished {
+                index,
+                exit,
+                bytes,
+                duration,
+            } => {
+                write!(
+                    f,
+                    "segment {index} on {exit}: {bytes} bytes in {duration:.2?}"
+                )
+            }
+            Self::SegmentFailed {
+                index,
+                exit,
+                attempt,
+                error,
+            } => {
+                write!(
+                    f,
+                    "segment {index} on {exit}: attempt {attempt} failed: {error}"
+                )
+            }
+        }
+    }
+}
+
+/// Cloneable handle to a progress callback; `Send + Sync`, so it can be
+/// shared across tokio tasks. All clones call the same callback.
+#[derive(Clone)]
+pub struct ProgressSink {
+    on_event: Arc<dyn Fn(&ProgressEvent) + Send + Sync>,
+}
+
+impl ProgressSink {
+    /// Wrap any event callback into a sink.
+    pub fn new<F>(on_event: F) -> Self
+    where
+        F: Fn(&ProgressEvent) + Send + Sync + 'static,
+    {
+        Self {
+            on_event: Arc::new(on_event),
+        }
+    }
+
+    /// A sink that discards every event.
+    #[must_use]
+    pub fn noop() -> Self {
+        Self::new(|_| ())
+    }
+
+    /// A sink that logs every event through `tracing`: started attempts
+    /// at debug, finished segments at info, failures at warn.
+    #[must_use]
+    pub fn tracing_log() -> Self {
+        Self::new(|event| match event {
+            ProgressEvent::SegmentStarted { .. } => tracing::debug!("{event}"),
+            ProgressEvent::SegmentFinished { .. } => tracing::info!("{event}"),
+            ProgressEvent::SegmentFailed { .. } => tracing::warn!("{event}"),
+        })
+    }
+
+    /// Combine two sinks: both receive every event.
+    #[must_use]
+    pub fn fanout(first: &Self, second: &Self) -> Self {
+        let first = first.clone();
+        let second = second.clone();
+        Self::new(move |event| {
+            first.on_event(event);
+            second.on_event(event);
+        })
+    }
+
+    /// Deliver one event to the underlying callback.
+    pub fn on_event(&self, event: &ProgressEvent) {
+        (self.on_event)(event);
+    }
+}
+
+impl fmt::Debug for ProgressSink {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ProgressSink").finish_non_exhaustive()
+    }
+}
+
+/// Counts the bytes of successfully finished segments across the whole
+/// fetch. Cheap to clone: every clone observes the same total.
+#[derive(Debug, Default, Clone)]
+pub struct BytesTracker {
+    bytes: Arc<AtomicU64>,
+}
+
+impl BytesTracker {
+    #[must_use]
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Total bytes reported as finished so far.
+    #[must_use]
+    pub fn total(&self) -> u64 {
+        self.bytes.load(Ordering::Relaxed)
+    }
+
+    /// A sink that feeds this tracker; the tracker stays queryable.
+    #[must_use]
+    pub fn sink(&self) -> ProgressSink {
+        let tracker = self.clone();
+        ProgressSink::new(move |event| tracker.record(event))
+    }
+
+    fn record(&self, event: &ProgressEvent) {
+        if let ProgressEvent::SegmentFinished { bytes, .. } = event {
+            self.bytes.fetch_add(*bytes, Ordering::Relaxed);
+        }
+    }
+}
@@ -10,12 +10,13 @@ use sha2::Digest;
 use tokio::io::AsyncWriteExt;

 use crate::error::FetchError;
-use crate::exec::{self, FetchReport, SegmentStats};
-use crate::exit::{ExitRegistry, LocalExit};
+use crate::exec::{self, FetchReport, RetryPolicy, SegmentStats};
+use crate::exit::{ExitRegistry, HttpExit, LocalExit, MeshExit};
 use crate::plan::ExitSpecs;
 use crate::probe::{self, ProbeInfo};
+use crate::progress::ProgressSink;
 use crate::types::{
-    FileSize, MinSegmentBytes, RangeSupport, SegmentsPerExit, Sha256Digest, TargetUrl,
+    Egress, FileSize, MinSegmentBytes, RangeSupport, SegmentsPerExit, Sha256Digest, TargetUrl,
 };

 /// All knobs for one fetch.
@@ -26,6 +27,8 @@ pub struct FetchOptions {
    pub segments_per_exit: SegmentsPerExit,
    pub min_segment: MinSegmentBytes,
    pub expected_sha256: Option<Sha256Digest>,
+    /// How often a segment is retried (preferably on another exit).
+    pub retries: RetryPolicy,
 }

 /// Run a full fetch and produce a report.
@@ -74,7 +77,17 @@ async fn run_ranged(
    tracing::info!("plan: {} segment(s)", plan.segments.len());

    let started = Instant::now();
-    let segment_stats: Vec<SegmentStats> = exec::run_segments(&plan, exits, parts_dir).await?;
+    let segment_stats: Vec<SegmentStats> = exec::run_segments_with_policy(
+        &plan,
+        exits,
+        parts_dir,
+        &options.retries,
+        &ProgressSink::tracing_log(),
+    )
+    .await?
+    .into_iter()
+    .map(std::convert::From::from)
+    .collect();
    let digest = exec::assemble(
        parts_dir,
        &options.output,
@@ -105,7 +118,7 @@ async fn run_whole(
    let exit_id = options
        .exits
        .first()
-        .map(|(id, _)| *id)
+        .map(|(id, _, _)| *id)
        .ok_or_else(|| FetchError::InvalidPlan("no exits given".to_string()))?;
    let client = exits.get(exit_id)?.client().clone();

@@ -163,11 +176,18 @@ async fn run_whole(
    })
 }

-/// Convenience: build a registry with one [`LocalExit`] per spec id.
-pub fn local_registry(specs: &ExitSpecs) -> Result<ExitRegistry, FetchError> {
+/// Build a registry from exit specs: `Egress::Direct` exits connect
+/// locally (own uplink), `Egress::Proxy` exits route through the given
+/// proxy — typically a neighbor's exit service over the Yggdrasil
+/// overlay.
+pub fn registry_from_specs(specs: &ExitSpecs) -> Result<ExitRegistry, FetchError> {
    let mut registry = ExitRegistry::empty();
-    for (id, _) in specs {
-        registry.register(std::sync::Arc::new(LocalExit::new(*id)?));
+    for (id, _, egress) in specs {
+        let exit: std::sync::Arc<dyn HttpExit> = match egress {
+            Egress::Direct => std::sync::Arc::new(LocalExit::new(*id)?),
+            Egress::Proxy(url) => std::sync::Arc::new(MeshExit::new(*id, url.clone())?),
+        };
+        registry.register(exit);
    }
    Ok(registry)
 }
@@ -230,3 +230,36 @@ pub enum RangeSupport {
    /// Probe was inconclusive; treat as unsupported (safe fallback).
    Unknown,
 }
+
+/// How a request physically leaves through an exit.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Egress {
+    /// The household's own uplink: plain outgoing HTTP, no detour.
+    Direct,
+    /// Through an HTTP proxy — e.g. a neighbor's exit service reachable
+    /// over the mesh overlay (typically bound to their Yggdrasil
+    /// address).
+    Proxy(reqwest::Url),
+}
+
+impl Egress {
+    /// Parse a proxy URL for [`Egress::Proxy`]; only `http`/`https`
+    /// upstreams are accepted.
+    pub fn proxy(input: &str) -> Result<Self, FetchError> {
+        let url =
+            reqwest::Url::parse(input).map_err(|_| FetchError::InvalidUrl(input.to_string()))?;
+        match url.scheme() {
+            "http" | "https" => Ok(Self::Proxy(url)),
+            _ => Err(FetchError::InvalidUrl(input.to_string())),
+        }
+    }
+}
+
+impl fmt::Display for Egress {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Direct => write!(f, "direct"),
+            Self::Proxy(url) => write!(f, "proxy:{url}"),
+        }
+    }
+}
@@ -1,5 +1,7 @@
 //! A minimal HTTP server for integration tests. Supports HEAD, GET with
-//! optional range handling (configurable), `Connection: close` semantics.
+//! optional range handling (configurable), simulated failures for range
+//! GETs (flaky/exhausted/non-retryable modes), and a broken exit whose
+//! client fails at transport level, for retry tests.

 #![forbid(unsafe_code)]
 // Tests are allowed to unwrap/expect; production code denies them via
@@ -8,8 +10,36 @@

 use std::io::{BufRead, BufReader, Write};
 use std::net::{SocketAddr, TcpListener, TcpStream};
+use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
 use std::thread::JoinHandle;
+use std::time::Duration;
+
+use sn_fetch::exit::HttpExit;
+use sn_fetch::types::ExitId;
+
+/// When the test origin should fail `Range` GETs (retry simulation).
+/// Only used by the retry test binary; allow dead_code for the others.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+#[allow(dead_code)]
+pub enum FailMode {
+    #[default]
+    Never,
+    /// Fail the first `n` range GETs with HTTP 500, then serve normally.
+    FirstN(u32),
+    /// Fail every range GET with HTTP 500.
+    Always,
+    /// Answer every range GET with the given status (e.g. 416 to
+    /// simulate a non-retryable range rejection).
+    AlwaysStatus(u16),
+}
+
+/// Knobs for the test origin.
+#[derive(Debug, Clone, Copy, Default)]
+#[allow(dead_code)]
+pub struct ServerConfig {
+    pub fail_mode: FailMode,
+}

 pub struct TestServer {
    pub addr: SocketAddr,
@@ -21,14 +51,25 @@ impl TestServer {
    /// server behaves like a range-hostile origin: no `Accept-Ranges`
    /// header, and `Range` requests are answered with the full body.
    pub fn start(data: Vec<u8>, ranges: bool) -> Self {
+        Self::start_with(data, ranges, ServerConfig::default())
+    }
+
+    /// Start with a custom [`ServerConfig`] (failure simulation).
+    pub fn start_with(data: Vec<u8>, ranges: bool, config: ServerConfig) -> Self {
        let data = Arc::new(data);
+        let config = Arc::new(config);
+        let range_get_count = Arc::new(AtomicU32::new(0));
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind in test");
-        let addr = listener.local_addr().expect("addr in test");
+        let addr = listener.local_addr().expect("local addr in test");
        let handle = std::thread::spawn(move || {
            for stream in listener.incoming().flatten() {
-                let data = Arc::clone(&data);
+                let (data, config, range_get_count) = (
+                    Arc::clone(&data),
+                    Arc::clone(&config),
+                    Arc::clone(&range_get_count),
+                );
                std::thread::spawn(move || {
-                    let _ = handle_conn(stream, &data, ranges);
+                    let _ = handle_conn(stream, &data, ranges, &config, &range_get_count);
                });
            }
        });
@@ -43,7 +84,13 @@ impl TestServer {
    }
 }

-fn handle_conn(stream: TcpStream, data: &[u8], ranges: bool) -> std::io::Result<()> {
+fn handle_conn(
+    stream: TcpStream,
+    data: &[u8],
+    ranges: bool,
+    config: &ServerConfig,
+    range_get_count: &AtomicU32,
+) -> std::io::Result<()> {
    let mut reader = BufReader::new(stream.try_clone()?);
    let mut request_line = String::new();
    reader.read_line(&mut request_line)?;
@@ -64,7 +111,6 @@ fn handle_conn(stream: TcpStream, data: &[u8], ranges: bool) -> std::io::Result<

    let mut stream = stream;
    let len = data.len() as u64;
-
    let method = request_line
        .split_whitespace()
        .next()
@@ -83,6 +129,10 @@ fn handle_conn(stream: TcpStream, data: &[u8], ranges: bool) -> std::io::Result<
            let range = range_header.as_deref().and_then(parse_range).flatten();
            match (ranges, range) {
                (true, Some((start, end))) => {
+                    if should_fail_range_get(config, range_get_count) {
+                        write_fail_response(&mut stream, config.fail_mode)?;
+                        return Ok(());
+                    }
                    let slice_len = end - start + 1;
                    write!(
                        stream,
@@ -111,6 +161,28 @@ fn handle_conn(stream: TcpStream, data: &[u8], ranges: bool) -> std::io::Result<
    Ok(())
 }

+/// Decide (and count) whether this range GET should fail. HEAD and
+/// non-range GETs are never counted nor failed.
+fn should_fail_range_get(config: &ServerConfig, range_get_count: &AtomicU32) -> bool {
+    match config.fail_mode {
+        FailMode::Never => false,
+        FailMode::Always | FailMode::AlwaysStatus(_) => true,
+        FailMode::FirstN(limit) => range_get_count.fetch_add(1, Ordering::Relaxed) < limit,
+    }
+}
+
+fn write_fail_response(stream: &mut TcpStream, mode: FailMode) -> std::io::Result<()> {
+    let status = match mode {
+        FailMode::AlwaysStatus(status) => status,
+        _ => 500,
+    };
+    write!(
+        stream,
+        "HTTP/1.1 {status} Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
+    )?;
+    stream.flush()
+}
+
 /// Parse `bytes=a-b` into inclusive `(start, end)`. Anything else
 /// (open-ended, malformed) is treated as absent.
 fn parse_range(value: &str) -> Option<Option<(u64, u64)>> {
@@ -124,3 +196,35 @@ fn parse_range(value: &str) -> Option<Option<(u64, u64)>> {
        Some(None)
    }
 }
+
+/// An exit whose client sends every request through a proxy that
+/// refuses connections — every request through this exit fails at
+/// transport level (retryable), as needed for exit-reassignment tests.
+#[allow(dead_code)]
+pub struct BrokenProxyExit {
+    id: ExitId,
+    client: reqwest::Client,
+}
+
+impl BrokenProxyExit {
+    #[allow(dead_code)]
+    pub fn new(id: ExitId) -> Self {
+        let proxy = reqwest::Proxy::http("http://127.0.0.1:1").expect("static proxy url parses");
+        let client = reqwest::Client::builder()
+            .proxy(proxy)
+            .connect_timeout(Duration::from_secs(10))
+            .build()
+            .expect("client builds in test");
+        Self { id, client }
+    }
+}
+
+impl HttpExit for BrokenProxyExit {
+    fn id(&self) -> ExitId {
+        self.id
+    }
+
+    fn client(&self) -> &reqwest::Client {
+        &self.client
+    }
+}
@@ -0,0 +1,151 @@
+//! End-to-end test: a `MeshExit` fetches through a real HTTP proxy.
+//!
+//! The fake proxy is a minimal HTTP forward proxy: it accepts absolute-URI
+//! requests (`GET http://origin/file.bin HTTP/1.1`), extracts the `Range`
+//! header and serves the requested slice from an in-memory buffer —
+//! exactly the shape of an `sn-exit` service bound to a neighbor's
+//! Yggdrasil address (minus the overlay transport, which is transparent
+//! to HTTP).
+
+#![forbid(unsafe_code)]
+#![allow(clippy::unwrap_used, clippy::expect_used)]
+
+use std::num::{NonZeroU32, NonZeroU64, NonZeroU8, NonZeroUsize};
+use std::sync::Arc;
+
+use sha2::{Digest, Sha256};
+use tempfile::TempDir;
+
+use sn_fetch::exit::{ExitRegistry, MeshExit};
+use sn_fetch::plan::{ExitSpecs, FetchPlan};
+use sn_fetch::types::{
+    Egress, ExitId, ExitWeight, FileSize, MinSegmentBytes, SegmentsPerExit, TargetUrl,
+};
+
+/// Minimal forward proxy serving one fixed object from memory.
+async fn spawn_proxy(data: Arc<Vec<u8>>) -> std::net::SocketAddr {
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        loop {
+            let Ok((mut socket, _)) = listener.accept().await else {
+                break;
+            };
+            let data = Arc::clone(&data);
+            tokio::spawn(async move {
+                // Read the request head.
+                let mut head = Vec::new();
+                let mut byte = [0u8; 1];
+                while !head.windows(4).any(|w| w == b"\r\n\r\n") {
+                    if socket.read(&mut byte).await.unwrap_or(0) == 0 {
+                        return;
+                    }
+                    head.push(byte[0]);
+                }
+                let head_text = String::from_utf8_lossy(&head);
+
+                // First line: `GET http://origin/path HTTP/1.1`.
+                let first_line = head_text.lines().next().unwrap_or_default();
+                if !first_line.starts_with("GET http://") {
+                    socket
+                        .write_all(b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\n\r\n")
+                        .await
+                        .ok();
+                    return;
+                }
+
+                // Extract the requested byte range (if any). Header
+                // names arrive lowercased from some clients.
+                let range = head_text.lines().find_map(|line| {
+                    let (name, value) = line.split_once(':')?;
+                    if !name.eq_ignore_ascii_case("range") {
+                        return None;
+                    }
+                    let value = value.trim().strip_prefix("bytes=")?;
+                    let (start, end) = value.trim().split_once('-')?;
+                    Some((start.parse::<u64>().ok()?, end.parse::<u64>().ok()?))
+                });
+
+                let (start, end_inclusive, status) = match range {
+                    Some((start, end)) => (start, end, "206 Partial Content"),
+                    None => (0, data.len() as u64 - 1, "200 OK"),
+                };
+                let slice = &data[start as usize..=(end_inclusive as usize).min(data.len() - 1)];
+                let response = format!(
+                    "HTTP/1.1 {status}\r\nContent-Length: {}\r\n\r\n",
+                    slice.len()
+                );
+                socket.write_all(response.as_bytes()).await.ok();
+                socket.write_all(slice).await.ok();
+                socket.shutdown().await.ok();
+            });
+        }
+    });
+    addr
+}
+
+#[tokio::test]
+async fn mesh_exit_downloads_through_the_proxy() {
+    let data: Vec<u8> = (0..64_000u32).map(|i| (i % 251) as u8).collect();
+    let data = Arc::new(data.clone());
+    let proxy_addr = spawn_proxy(Arc::clone(&data)).await;
+
+    let proxy_url = format!("http://{proxy_addr}");
+    let specs: ExitSpecs = vec![(
+        ExitId::new(NonZeroU8::new(1).expect("1 is non-zero")),
+        ExitWeight::new(NonZeroU32::new(1).expect("1 is non-zero")),
+        Egress::proxy(&proxy_url).expect("proxy url is valid"),
+    )];
+
+    // The origin URL is fictional — the proxy serves the content; this
+    // proves the request really went through the proxy, not straight to
+    // an origin server.
+    let target = TargetUrl::parse("http://origin.invalid/file.bin").expect("valid url");
+    let file_size = FileSize::try_new(data.len() as u64).expect("non-zero in test");
+
+    let plan = FetchPlan::build(
+        target.clone(),
+        file_size,
+        &specs,
+        SegmentsPerExit::new(NonZeroUsize::new(2).expect("2 is non-zero")),
+        MinSegmentBytes::new(NonZeroU64::new(1024).expect("1024 is non-zero")),
+    )
+    .expect("plan builds");
+    plan.validate().expect("plan validates");
+
+    let mut registry = ExitRegistry::empty();
+    registry.register(Arc::new(
+        MeshExit::new(
+            ExitId::new(NonZeroU8::new(1).expect("1 is non-zero")),
+            reqwest::Url::parse(&proxy_url).expect("proxy url is valid"),
+        )
+        .expect("mesh exit builds"),
+    ));
+
+    let parts = TempDir::new().expect("tempdir in test");
+    let stats = sn_fetch::exec::run_segments(&plan, &registry, parts.path())
+        .await
+        .expect("segments succeed through the proxy");
+    let total: u64 = stats.iter().map(|s| s.bytes).sum();
+    assert_eq!(total, data.len() as u64);
+
+    let output = parts.path().join("mesh.bin");
+    let expected_digest =
+        sn_fetch::types::Sha256Digest::from_bytes(Sha256::digest(data.as_ref().as_slice()).into());
+    let digest = sn_fetch::exec::assemble(parts.path(), &output, file_size, Some(&expected_digest))
+        .await
+        .expect("assembly verifies");
+    assert_eq!(digest, expected_digest);
+}
+
+/// The proxy's address (IPv4, arbitrary port) is not a valid proxy URL
+/// unless it has an http scheme — Egress::proxy must reject junk.
+#[test]
+fn egress_rejects_non_http_proxy_schemes() {
+    assert!(Egress::proxy("ftp://proxy:2121").is_err());
+    assert!(Egress::proxy("not a url").is_err());
+    assert!(Egress::proxy("http://127.0.0.1:8080").is_ok());
+    assert!(Egress::proxy("https://proxy.example:3128").is_ok());
+}
@@ -0,0 +1,428 @@
+//! Retry behaviour against a local HTTP server: a flaky first attempt
+//! (retried, reassigned to another exit), a consistently failing exit
+//! (reassigned), retry exhaustion (whole fetch fails, no output file)
+//! and a non-retryable status (immediate abort).
+
+#![forbid(unsafe_code)]
+// Tests are allowed to unwrap/expect; production code denies them via
+// workspace lints (server/Cargo.toml).
+#![allow(clippy::unwrap_used, clippy::expect_used)]
+
+mod common;
+
+use std::num::{NonZeroU32, NonZeroU64, NonZeroU8, NonZeroUsize};
+use std::path::Path;
+use std::sync::{Arc, Mutex};
+
+use sha2::{Digest, Sha256};
+use tempfile::TempDir;
+
+use sn_fetch::error::FetchError;
+use sn_fetch::exec::{self, RetryPolicy};
+use sn_fetch::exit::{ExitRegistry, LocalExit};
+use sn_fetch::plan::{ExitSpecs, FetchPlan};
+use sn_fetch::progress::{BytesTracker, ProgressEvent, ProgressSink};
+use sn_fetch::types::{
+    Egress, ExitId, ExitWeight, FileSize, MinSegmentBytes, SegmentsPerExit, TargetUrl,
+};
+
+use common::{BrokenProxyExit, FailMode, ServerConfig, TestServer};
+
+fn spec(id: u8) -> (ExitId, ExitWeight, Egress) {
+    (
+        ExitId::new(NonZeroU8::new(id).expect("test ids are 1..=255")),
+        ExitWeight::new(NonZeroU32::new(1).expect("1 is non-zero")),
+        Egress::Direct,
+    )
+}
+
+fn exit_id(id: u8) -> ExitId {
+    ExitId::new(NonZeroU8::new(id).expect("test ids are 1..=255"))
+}
+
+fn local_registry(specs: &ExitSpecs) -> ExitRegistry {
+    let mut registry = ExitRegistry::empty();
+    for (id, _, _) in specs {
+        registry.register(Arc::new(LocalExit::new(*id).expect("local exit")));
+    }
+    registry
+}
+
+/// Single-exit plan + registry, for exhaustion tests.
+fn single_exit_setup(
+    url: &TargetUrl,
+    file_size: u64,
+    min_segment: u64,
+) -> (FetchPlan, ExitRegistry) {
+    let specs: ExitSpecs = vec![spec(1)];
+    let plan = FetchPlan::build(
+        url.clone(),
+        FileSize::try_new(file_size).expect("non-zero in test"),
+        &specs,
+        SegmentsPerExit::new(NonZeroUsize::MIN),
+        MinSegmentBytes::new(NonZeroU64::new(min_segment).expect("non-zero in test")),
+    )
+    .expect("plan builds");
+    let registry = local_registry(&specs);
+    (plan, registry)
+}
+
+type RecordedEvents = Arc<Mutex<Vec<ProgressEvent>>>;
+
+fn recording_sink() -> (ProgressSink, RecordedEvents) {
+    let events: RecordedEvents = Arc::new(Mutex::new(Vec::new()));
+    let sink = ProgressSink::new({
+        let events = Arc::clone(&events);
+        move |event| {
+            events
+                .lock()
+                .expect("mutex poisoned in test")
+                .push(event.clone())
+        }
+    });
+    (sink, events)
+}
+
+fn failed_events(events: &RecordedEvents) -> Vec<(u32, u32, String)> {
+    events
+        .lock()
+        .expect("mutex poisoned in test")
+        .iter()
+        .filter_map(|event| match event {
+            ProgressEvent::SegmentFailed {
+                index,
+                exit: _,
+                attempt,
+                error,
+            } => Some((index.value(), *attempt, error.clone())),
+            _ => None,
+        })
+        .collect()
+}
+
+/// Names of leftover part files in `parts_dir`.
+fn part_files(dir: &Path) -> Vec<String> {
+    std::fs::read_dir(dir)
+        .expect("parts dir readable in test")
+        .flatten()
+        .filter(|entry| entry.file_name().to_string_lossy().ends_with(".part"))
+        .map(|entry| entry.file_name().to_string_lossy().into_owned())
+        .collect()
+}
+
+#[tokio::test]
+async fn flaky_segment_retried_on_other_exit() {
+    let data: Vec<u8> = (0..(2 * 1024 * 1024)).map(|i| (i % 253) as u8).collect();
+    let server = TestServer::start_with(
+        data.clone(),
+        true,
+        ServerConfig {
+            fail_mode: FailMode::FirstN(1),
+        },
+    );
+    let url = TargetUrl::parse(&server.url("/file.bin")).expect("valid url");
+
+    let specs: ExitSpecs = vec![spec(1), spec(2)];
+    let plan = FetchPlan::build(
+        url,
+        FileSize::try_new(data.len() as u64).expect("non-zero in test"),
+        &specs,
+        SegmentsPerExit::new(NonZeroUsize::new(2).expect("2 is non-zero")),
+        MinSegmentBytes::new(NonZeroU64::new(1024).expect("1024 is non-zero")),
+    )
+    .expect("plan builds");
+    let registry = local_registry(&specs);
+
+    let parts = TempDir::new().expect("tempdir in test");
+    let (progress, events) = recording_sink();
+    let tracker = BytesTracker::new();
+    let combined = ProgressSink::fanout(&progress, &tracker.sink());
+
+    let stats = exec::run_segments_with_policy(
+        &plan,
+        &registry,
+        parts.path(),
+        &RetryPolicy::default(),
+        &combined,
+    )
+    .await
+    .expect("fetch succeeds despite one flaky attempt");
+
+    assert_eq!(stats.len(), plan.segments.len());
+    let total: u64 = stats.iter().map(|s| s.bytes).sum();
+    assert_eq!(total, data.len() as u64);
+    assert_eq!(
+        tracker.total(),
+        data.len() as u64,
+        "tracker must see all finished bytes"
+    );
+
+    let retried: Vec<_> = stats.iter().filter(|s| s.attempts > 1).collect();
+    assert_eq!(retried.len(), 1, "exactly one segment needed a retry");
+    let retried = retried[0];
+    let planned_exit = plan
+        .segments
+        .iter()
+        .find(|seg| seg.index == retried.index)
+        .expect("planned segment")
+        .exit;
+    assert_ne!(retried.exit, planned_exit, "retry must use another exit");
+    assert_eq!(retried.attempts, 2);
+
+    let started: Vec<(u8, u32)> = events
+        .lock()
+        .expect("mutex poisoned in test")
+        .iter()
+        .filter_map(|event| match event {
+            ProgressEvent::SegmentStarted {
+                index,
+                exit,
+                attempt,
+            } if index == &retried.index => Some((exit.value(), *attempt)),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(
+        started,
+        vec![(planned_exit.value(), 1), (retried.exit.value(), 2)],
+        "the retried segment must start once per attempt, on different exits"
+    );
+    let failed = failed_events(&events);
+    assert_eq!(
+        failed,
+        vec![(retried.index.value(), 1, failed[0].2.clone())],
+        "exactly one failed attempt, recorded on attempt 1"
+    );
+    assert!(
+        failed[0].2.contains("500"),
+        "failure reason should mention the 500 status: {}",
+        failed[0].2
+    );
+}
+
+#[tokio::test]
+async fn exhausted_retries_fail_without_output() {
+    let server = TestServer::start_with(
+        vec![7u8; 32 * 1024],
+        true,
+        ServerConfig {
+            fail_mode: FailMode::Always,
+        },
+    );
+    let url = TargetUrl::parse(&server.url("/file.bin")).expect("valid url");
+    let (plan, registry) = single_exit_setup(&url, 32 * 1024, 32 * 1024);
+    assert_eq!(plan.segments.len(), 1);
+
+    let parts = TempDir::new().expect("tempdir in test");
+    let (progress, events) = recording_sink();
+    let tracker = BytesTracker::new();
+    let combined = ProgressSink::fanout(&progress, &tracker.sink());
+
+    let result = exec::run_segments_with_policy(
+        &plan,
+        &registry,
+        parts.path(),
+        &RetryPolicy::default(),
+        &combined,
+    )
+    .await;
+
+    let err = result.expect_err("fetch must fail when every attempt fails");
+    assert!(
+        matches!(
+            err,
+            FetchError::SegmentAttemptsExhausted { attempts: 3, .. }
+        ),
+        "unexpected error: {err}"
+    );
+    assert_eq!(
+        tracker.total(),
+        0,
+        "nothing finished, tracker stays at zero"
+    );
+
+    let failed: Vec<u32> = failed_events(&events)
+        .into_iter()
+        .map(|(_, attempt, error)| {
+            assert!(error.contains("500"), "failure reason: {error}");
+            attempt
+        })
+        .collect();
+    assert_eq!(failed, vec![1, 2, 3], "default policy means three attempts");
+
+    assert!(
+        part_files(parts.path()).is_empty(),
+        "failed part files must be cleaned up between attempts"
+    );
+    let output = parts.path().join("output.bin");
+    assert!(
+        !output.exists(),
+        "no output file may be created for a failed fetch"
+    );
+}
+
+#[tokio::test]
+async fn single_attempt_policy_fails_fast() {
+    let server = TestServer::start_with(
+        vec![5u8; 16 * 1024],
+        true,
+        ServerConfig {
+            fail_mode: FailMode::Always,
+        },
+    );
+    let url = TargetUrl::parse(&server.url("/file.bin")).expect("valid url");
+    let (plan, registry) = single_exit_setup(&url, 16 * 1024, 16 * 1024);
+
+    let parts = TempDir::new().expect("tempdir in test");
+    let (progress, events) = recording_sink();
+    let policy = RetryPolicy::new(NonZeroU32::new(1).expect("1 is non-zero"));
+
+    let result =
+        exec::run_segments_with_policy(&plan, &registry, parts.path(), &policy, &progress).await;
+
+    let err = result.expect_err("fetch must fail after the single attempt");
+    assert!(
+        matches!(
+            err,
+            FetchError::SegmentAttemptsExhausted { attempts: 1, .. }
+        ),
+        "unexpected error: {err}"
+    );
+    assert_eq!(failed_events(&events).len(), 1);
+    assert!(part_files(parts.path()).is_empty());
+}
+
+#[tokio::test]
+async fn failing_exit_reassigned_to_working_exit() {
+    let data: Vec<u8> = (0..(1024 * 1024)).map(|i| (i % 241) as u8).collect();
+    let expected_digest = sn_fetch::types::Sha256Digest::from_bytes(Sha256::digest(&data).into());
+    let server = TestServer::start(data.clone(), true);
+    let url = TargetUrl::parse(&server.url("/file.bin")).expect("valid url");
+
+    let specs: ExitSpecs = vec![spec(1), spec(2)];
+    let plan = FetchPlan::build(
+        url,
+        FileSize::try_new(data.len() as u64).expect("non-zero in test"),
+        &specs,
+        SegmentsPerExit::new(NonZeroUsize::MIN),
+        MinSegmentBytes::new(NonZeroU64::new(1024).expect("1024 is non-zero")),
+    )
+    .expect("plan builds");
+
+    let mut registry = ExitRegistry::empty();
+    registry.register(Arc::new(BrokenProxyExit::new(exit_id(1))));
+    registry.register(Arc::new(LocalExit::new(exit_id(2)).expect("local exit")));
+
+    let parts = TempDir::new().expect("tempdir in test");
+    let (progress, events) = recording_sink();
+
+    let stats = exec::run_segments_with_policy(
+        &plan,
+        &registry,
+        parts.path(),
+        &RetryPolicy::default(),
+        &progress,
+    )
+    .await
+    .expect("fetch succeeds by reassigning off the broken exit");
+
+    for stat in &stats {
+        let planned = plan
+            .segments
+            .iter()
+            .find(|seg| seg.index == stat.index)
+            .expect("planned segment");
+        if planned.exit.value() == 1 {
+            assert_eq!(
+                stat.attempts, 2,
+                "segment planned on the broken exit needs exactly one retry"
+            );
+            assert_eq!(stat.exit.value(), 2, "retry must land on the working exit");
+        } else {
+            assert_eq!(stat.attempts, 1);
+            assert_eq!(stat.exit.value(), 2);
+        }
+    }
+
+    let started_exits: Vec<u8> = events
+        .lock()
+        .expect("mutex poisoned in test")
+        .iter()
+        .filter_map(|event| match event {
+            ProgressEvent::SegmentStarted { index, exit, .. } if index.value() == 0 => {
+                Some(exit.value())
+            }
+            _ => None,
+        })
+        .collect();
+    assert_eq!(
+        started_exits,
+        vec![1, 2],
+        "segment 0 must be tried on the broken exit first, then reassigned"
+    );
+
+    let output = parts.path().join("out.bin");
+    let digest = exec::assemble(
+        parts.path(),
+        &output,
+        FileSize::try_new(data.len() as u64).expect("non-zero in test"),
+        Some(&expected_digest),
+    )
+    .await
+    .expect("assembly verifies");
+    assert_eq!(digest, expected_digest);
+}
+
+#[tokio::test]
+async fn non_retryable_status_aborts_immediately() {
+    let server = TestServer::start_with(
+        vec![3u8; 32 * 1024],
+        true,
+        ServerConfig {
+            fail_mode: FailMode::AlwaysStatus(416),
+        },
+    );
+    let url = TargetUrl::parse(&server.url("/file.bin")).expect("valid url");
+    let (plan, registry) = single_exit_setup(&url, 32 * 1024, 32 * 1024);
+
+    let parts = TempDir::new().expect("tempdir in test");
+    let (progress, events) = recording_sink();
+
+    let result = exec::run_segments_with_policy(
+        &plan,
+        &registry,
+        parts.path(),
+        &RetryPolicy::default(),
+        &progress,
+    )
+    .await;
+
+    let err = result.expect_err("416 must abort the fetch");
+    assert!(
+        matches!(
+            err,
+            FetchError::Http(ref http_err)
+                if http_err.status() == Some(reqwest::StatusCode::RANGE_NOT_SATISFIABLE)
+        ),
+        "the raw 416 must surface unchanged, got: {err}"
+    );
+
+    // Non-retryable: one started attempt, one failure event, no retry.
+    let guard = events.lock().expect("mutex poisoned in test");
+    let started = guard
+        .iter()
+        .filter(|event| matches!(event, ProgressEvent::SegmentStarted { .. }))
+        .count();
+    assert_eq!(started, 1, "non-retryable failure must not be retried");
+    let failed = guard
+        .iter()
+        .filter(|event| matches!(event, ProgressEvent::SegmentFailed { .. }))
+        .count();
+    assert_eq!(failed, 1);
+    drop(guard);
+
+    assert!(
+        part_files(parts.path()).is_empty(),
+        "failed part files must be cleaned up"
+    );
+}
@@ -21,8 +21,8 @@ use sn_fetch::exit::LocalExit;
 use sn_fetch::plan::{ExitSpecs, FetchPlan};
 use sn_fetch::probe;
 use sn_fetch::types::{
-    ByteRange, ExitId, ExitWeight, FileSize, MinSegmentBytes, RangeSupport, SegmentsPerExit,
-    Sha256Digest,
+    ByteRange, Egress, ExitId, ExitWeight, FileSize, MinSegmentBytes, RangeSupport,
+    SegmentsPerExit, Sha256Digest,
 };

 fn exit_specs(n: u8) -> ExitSpecs {
@@ -31,6 +31,7 @@ fn exit_specs(n: u8) -> ExitSpecs {
            (
                ExitId::new(NonZeroU8::new(i).expect("test ids are 1..=255")),
                ExitWeight::new(NonZeroU32::new(1).expect("1 is non-zero")),
+                Egress::Direct,
            )
        })
        .collect()
@@ -68,7 +69,7 @@ async fn segmented_download_reassembles_and_hashes() {
    plan.validate().expect("plan validates");

    let mut registry = sn_fetch::exit::ExitRegistry::empty();
-    for id in specs.iter().map(|(id, _)| *id) {
+    for id in specs.iter().map(|(id, _, _)| *id) {
        registry.register(Arc::new(LocalExit::new(id).expect("local exit")));
    }

@@ -109,8 +110,9 @@ async fn range_hostile_server_falls_back_to_single_stream() {
        segments_per_exit: SegmentsPerExit::new(NonZeroUsize::new(1).expect("1 is non-zero")),
        min_segment: MinSegmentBytes::new(NonZeroU64::new(1024).expect("1024 is non-zero")),
        expected_sha256: None,
+        retries: sn_fetch::exec::RetryPolicy::default(),
    };
-    let registry = sn_fetch::runner::local_registry(&specs).expect("registry");
+    let registry = sn_fetch::runner::registry_from_specs(&specs).expect("registry");
    let parts = TempDir::new().expect("tempdir in test");

    let report = sn_fetch::runner::run(&target, &options, &registry, parts.path())
@@ -131,10 +133,12 @@ fn plan_is_contiguous_and_weighted() {
        (
            ExitId::new(NonZeroU8::new(1).expect("1 is non-zero")),
            ExitWeight::new(NonZeroU32::new(3).expect("3 is non-zero")),
+            Egress::Direct,
        ),
        (
            ExitId::new(NonZeroU8::new(2).expect("2 is non-zero")),
            ExitWeight::new(NonZeroU32::new(1).expect("1 is non-zero")),
+            Egress::Direct,
        ),
    ];

@@ -0,0 +1,15 @@
+[package]
+name = "sn-node"
+description = "SpiderNet node helper: Yggdrasil overlay config generation and status queries"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+tokio = { workspace = true }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+clap = { workspace = true }
+
+[lints]
+workspace = true
@@ -0,0 +1,343 @@
+//! Yggdrasil configuration generation for a `SpiderNet` node.
+//!
+//! Yggdrasil's own `-genconf` output is used as the base (so we never
+//! hard-code version-specific defaults); the `SpiderNet` settings are then
+//! patched into that JSON base as validated, typed values. The result is
+//! written as JSON — Yggdrasil accepts JSON config via `-useconffile`.
+
+use std::path::{Path, PathBuf};
+
+use crate::error::NodeError;
+use crate::yggdrasil::AdminEndpoint;
+
+/// A validated Yggdrasil peering URI (`tcp://`, `tls://`, `quic://` or
+/// `socks://`).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PeerUri(String);
+
+impl PeerUri {
+    pub fn parse(input: &str) -> Result<Self, NodeError> {
+        Ok(Self(validate_scheme(input, "peering")?))
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+/// A validated listener URI a node binds for incoming peerings
+/// (`tcp://[::]:1337` style).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ListenUri(String);
+
+impl ListenUri {
+    pub fn parse(input: &str) -> Result<Self, NodeError> {
+        let trimmed = input.trim();
+        if trimmed.starts_with("socks://") {
+            return Err(NodeError::InvalidUri {
+                uri: input.to_string(),
+                hint: "listeners bind an address: use tcp://, tls:// or quic://",
+            });
+        }
+        Ok(Self(validate_scheme(trimmed, "listen")?))
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+const PEER_SCHEMES: [&str; 4] = ["tcp://", "tls://", "quic://", "socks://"];
+
+fn validate_scheme(input: &str, kind: &'static str) -> Result<String, NodeError> {
+    let trimmed = input.trim();
+    let scheme = PEER_SCHEMES
+        .iter()
+        .find(|scheme| trimmed.starts_with(**scheme));
+    match scheme {
+        Some(scheme) if trimmed.len() > scheme.len() => Ok(trimmed.to_string()),
+        _ => Err(NodeError::InvalidUri {
+            uri: input.to_string(),
+            hint: match kind {
+                "peering" => "peer URIs must start with tcp://, tls://, quic:// or socks://",
+                _ => "listen URIs must start with tcp://, tls:// or quic://",
+            },
+        }),
+    }
+}
+
+/// A validated Yggdrasil public key: 64 hex characters.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PublicKeyHex(String);
+
+impl PublicKeyHex {
+    pub fn parse(input: &str) -> Result<Self, NodeError> {
+        let trimmed = input.trim().to_lowercase();
+        let valid = trimmed.len() == 64 && trimmed.chars().all(|c| c.is_ascii_hexdigit());
+        if valid {
+            Ok(Self(trimmed))
+        } else {
+            Err(NodeError::InvalidPublicKey(input.to_string()))
+        }
+    }
+
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+/// Which TUN interface Yggdrasil should use.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum IfName {
+    /// `auto` — let Yggdrasil pick (default).
+    Auto,
+    /// `none` — headless router-only mode, no TUN device.
+    Headless,
+    Named(String),
+}
+
+/// Multicast (same-LAN) auto-peering mode.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum MulticastMode {
+    /// Leave Yggdrasil's generated default untouched.
+    PlatformDefault,
+    /// Disable multicast peering entirely.
+    Disabled,
+    /// Explicit interface entries (regex-matched).
+    Interfaces(Vec<MulticastInterface>),
+}
+
+/// One multicast peering entry, mirroring Yggdrasil's JSON shape.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct MulticastInterface {
+    /// Interface name regex, e.g. `eth0` or `.*`.
+    pub regex: String,
+    pub beacon: bool,
+    pub listen: bool,
+    pub port: u16,
+}
+
+/// Everything `SpiderNet` wants to control in the Yggdrasil config.
+/// Unlisted fields keep the values Yggdrasil's `-genconf` produced.
+#[derive(Debug, Clone, Default)]
+pub struct NodeSettings {
+    pub peers: Vec<PeerUri>,
+    pub listen: Vec<ListenUri>,
+    pub multicast: Option<MulticastMode>,
+    pub admin_listen: Option<AdminEndpoint>,
+    pub if_name: Option<IfName>,
+    /// Neighbor keys allowed to peer with us; empty means "allow all".
+    pub allowed_public_keys: Vec<PublicKeyHex>,
+    pub private_key_path: Option<PathBuf>,
+}
+
+/// Run `yggdrasil -genconf -json` and patch the `SpiderNet` settings into
+/// the generated base config. Returns the pretty-printed JSON config.
+pub fn generate_config(binary: &str, settings: &NodeSettings) -> Result<String, NodeError> {
+    let output = std::process::Command::new(binary)
+        .arg("-genconf")
+        .arg("-json")
+        .output()?;
+    if !output.status.success() {
+        let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
+        return Err(NodeError::YggdrasilBinary {
+            binary: binary.to_string(),
+            stderr,
+        });
+    }
+    let mut base: serde_json::Value = serde_json::from_slice(&output.stdout)?;
+    apply_to_base_config(settings, &mut base)?;
+    Ok(serde_json::to_string_pretty(&base)?)
+}
+
+/// Patch the `SpiderNet` settings into an existing (parsed) Yggdrasil base
+/// config. Pure function so it can be tested without the binary.
+pub fn apply_to_base_config(
+    settings: &NodeSettings,
+    base: &mut serde_json::Value,
+) -> Result<(), NodeError> {
+    let obj = base
+        .as_object_mut()
+        .ok_or(NodeError::UnexpectedConfigShape)?;
+
+    if !settings.peers.is_empty() {
+        let peers: Vec<&str> = settings.peers.iter().map(PeerUri::as_str).collect();
+        obj.insert("Peers".to_string(), serde_json::json!(peers));
+    }
+    if !settings.listen.is_empty() {
+        let listen: Vec<&str> = settings.listen.iter().map(ListenUri::as_str).collect();
+        obj.insert("Listen".to_string(), serde_json::json!(listen));
+    }
+    match settings.multicast.as_ref() {
+        None | Some(MulticastMode::PlatformDefault) => {}
+        Some(MulticastMode::Disabled) => {
+            obj.insert("MulticastInterfaces".to_string(), serde_json::json!([]));
+        }
+        Some(MulticastMode::Interfaces(interfaces)) => {
+            let entries: Vec<serde_json::Value> = interfaces
+                .iter()
+                .map(|iface| {
+                    serde_json::json!({
+                        "Beacon": iface.beacon,
+                        "Listen": iface.listen,
+                        "Port": iface.port,
+                        "Regex": iface.regex,
+                    })
+                })
+                .collect();
+            obj.insert(
+                "MulticastInterfaces".to_string(),
+                serde_json::json!(entries),
+            );
+        }
+    }
+    if let Some(admin_listen) = settings.admin_listen.as_ref() {
+        obj.insert(
+            "AdminListen".to_string(),
+            serde_json::json!(admin_listen.config_value()),
+        );
+    }
+    if let Some(if_name) = settings.if_name.as_ref() {
+        let value = match if_name {
+            IfName::Auto => "auto".to_string(),
+            IfName::Headless => "none".to_string(),
+            IfName::Named(name) => name.clone(),
+        };
+        obj.insert("IfName".to_string(), serde_json::json!(value));
+    }
+    if !settings.allowed_public_keys.is_empty() {
+        let keys: Vec<&str> = settings
+            .allowed_public_keys
+            .iter()
+            .map(PublicKeyHex::as_str)
+            .collect();
+        obj.insert("AllowedPublicKeys".to_string(), serde_json::json!(keys));
+    }
+    if let Some(private_key_path) = settings.private_key_path.as_ref() {
+        obj.insert(
+            "PrivateKeyPath".to_string(),
+            serde_json::json!(private_key_path.display().to_string()),
+        );
+    }
+    Ok(())
+}
+
+/// Write the config to `path` and restrict permissions to the owner
+/// (the file contains the node's private key — it must never be readable
+/// by other users, and it must never be committed to any repository).
+pub fn write_config(path: &Path, json: &str) -> Result<(), NodeError> {
+    use std::io::Write as _;
+
+    let mut file = std::fs::File::create(path)?;
+    file.write_all(json.as_bytes())?;
+    file.flush()?;
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+        std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?;
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::unwrap_used, clippy::expect_used)]
+    use super::*;
+
+    #[test]
+    fn peer_uri_accepts_supported_schemes() {
+        for uri in [
+            "tcp://host:1337",
+            "tls://host:1337",
+            "quic://host:1337",
+            "socks://p:1/h:2",
+        ] {
+            assert!(PeerUri::parse(uri).is_ok(), "{uri} should be valid");
+        }
+    }
+
+    #[test]
+    fn peer_uri_rejects_bare_or_unknown_schemes() {
+        for uri in ["host:1337", "http://host:80", "tcp://", ""] {
+            assert!(PeerUri::parse(uri).is_err(), "{uri} should be invalid");
+        }
+    }
+
+    #[test]
+    fn listen_uri_rejects_socks() {
+        assert!(ListenUri::parse("socks://[::]:1337").is_err());
+        assert!(ListenUri::parse("tcp://[::]:1337").is_ok());
+    }
+
+    #[test]
+    fn public_key_requires_64_hex_chars() {
+        assert!(PublicKeyHex::parse(&"a".repeat(64)).is_ok());
+        assert!(PublicKeyHex::parse(&"a".repeat(63)).is_err());
+        assert!(PublicKeyHex::parse(&"z".repeat(64)).is_err());
+    }
+
+    #[test]
+    fn admin_endpoint_parse_and_roundtrip() {
+        let tcp = AdminEndpoint::parse("tcp://localhost:9001").unwrap();
+        assert_eq!(
+            tcp,
+            AdminEndpoint::Tcp {
+                host: "localhost".into(),
+                port: 9001
+            }
+        );
+        assert_eq!(tcp.config_value(), "tcp://localhost:9001");
+
+        let unix = AdminEndpoint::parse("unix:///var/run/yggdrasil.sock").unwrap();
+        assert_eq!(unix.config_value(), "unix:///var/run/yggdrasil.sock");
+
+        assert!(AdminEndpoint::parse("http://localhost:9001").is_err());
+        assert!(AdminEndpoint::parse("tcp://localhost").is_err());
+    }
+
+    #[test]
+    fn patches_land_in_base_config() {
+        let mut base: serde_json::Value =
+            serde_json::from_str(r#"{"PrivateKey": "secret", "Peers": [], "Listen": []}"#).unwrap();
+        let settings = NodeSettings {
+            peers: vec![PeerUri::parse("tcp://[200:1234::1]:1337").unwrap()],
+            listen: vec![ListenUri::parse("tls://[::]:1337").unwrap()],
+            multicast: Some(MulticastMode::Disabled),
+            admin_listen: Some(AdminEndpoint::parse("tcp://localhost:9001").unwrap()),
+            if_name: Some(IfName::Headless),
+            allowed_public_keys: vec![PublicKeyHex::parse(&"ab".repeat(32)).unwrap()],
+            private_key_path: Some(PathBuf::from("/etc/spidernet/ygg.key")),
+        };
+        apply_to_base_config(&settings, &mut base).unwrap();
+
+        assert_eq!(base["Peers"].as_array().unwrap().len(), 1);
+        assert_eq!(base["Listen"][0], "tls://[::]:1337");
+        assert_eq!(base["MulticastInterfaces"].as_array().unwrap().len(), 0);
+        assert_eq!(base["AdminListen"], "tcp://localhost:9001");
+        assert_eq!(base["IfName"], "none");
+        assert_eq!(base["AllowedPublicKeys"][0], "ab".repeat(32));
+        assert_eq!(base["PrivateKeyPath"], "/etc/spidernet/ygg.key");
+        // The base's own keys survive the patch.
+        assert_eq!(base["PrivateKey"], "secret");
+    }
+
+    #[test]
+    fn multicast_interfaces_are_serialized_in_yggdrasil_shape() {
+        let mut base: serde_json::Value = serde_json::from_str("{}").unwrap();
+        let settings = NodeSettings {
+            multicast: Some(MulticastMode::Interfaces(vec![MulticastInterface {
+                regex: "eth0".to_string(),
+                beacon: true,
+                listen: true,
+                port: 1337,
+            }])),
+            ..NodeSettings::default()
+        };
+        apply_to_base_config(&settings, &mut base).unwrap();
+        let entry = &base["MulticastInterfaces"][0];
+        assert_eq!(entry["Regex"], "eth0");
+        assert_eq!(entry["Beacon"], true);
+        assert_eq!(entry["Listen"], true);
+        assert_eq!(entry["Port"], 1337);
+    }
+}
@@ -0,0 +1,34 @@
+//! Error type for all sn-node operations.
+
+#[derive(Debug, thiserror::Error)]
+pub enum NodeError {
+    #[error("io error: {0}")]
+    Io(#[from] std::io::Error),
+
+    #[error("json error: {0}")]
+    Json(#[from] serde_json::Error),
+
+    #[error("yggdrasil binary `{binary}` failed: {stderr}")]
+    YggdrasilBinary { binary: String, stderr: String },
+
+    #[error("invalid uri `{uri}`: {hint}")]
+    InvalidUri { uri: String, hint: &'static str },
+
+    #[error("invalid admin endpoint `{0}`: expected `tcp://host:port` or `unix:///path`")]
+    InvalidAdminEndpoint(String),
+
+    #[error("invalid public key `{0}`: expected 64 hex characters")]
+    InvalidPublicKey(String),
+
+    #[error("generated config base has unexpected shape (expected a JSON object)")]
+    UnexpectedConfigShape,
+
+    #[error("admin protocol violation: {0}")]
+    AdminProtocol(String),
+
+    #[error("admin socket answered with an error: {0}")]
+    AdminFailed(String),
+
+    #[error("admin socket operation timed out")]
+    Timeout,
+}
@@ -0,0 +1,28 @@
+//! sn-node — `SpiderNet` node helper for the Yggdrasil overlay.
+//!
+//! `SpiderNet` is software for neighborhood fiber meshes: households peer
+//! their internet uplinks and pool bandwidth. This crate is the node-side
+//! glue for the overlay layer:
+//!
+//! - [`config`]: generate a Yggdrasil configuration file from validated,
+//!   typed `SpiderNet` settings (peers, listeners, admin socket, allowed
+//!   neighbor keys).
+//! - [`yggdrasil`]: client for the Yggdrasil admin socket (JSON over
+//!   TCP/unix, default `localhost:9001`) to query node status (`getSelf`,
+//!   `getPeers`).
+//!
+//! Yggdrasil provides encrypted end-to-end routing between households.
+//! It is **not** an anonymity network, and `SpiderNet` never claims to
+//! provide anonymity.
+//!
+//! `SpiderNet` software carries no warranty and accepts no responsibility
+//! for its usage or for transported content (AGPL-3.0, sections 15/16).
+
+#![forbid(unsafe_code)]
+#![warn(clippy::pedantic)]
+// PoC scope: doc-annotation noise, not correctness. Revisit at release.
+#![allow(clippy::must_use_candidate, clippy::missing_errors_doc)]
+
+pub mod config;
+pub mod error;
+pub mod yggdrasil;
@@ -0,0 +1,229 @@
+//! CLI entry point for sn-node.
+//!
+//! KEINE GARANTIE — this software is provided "as is" without warranty of
+//! any kind (AGPL-3.0, sections 15/16). Users are solely responsible for
+//! what they transmit or receive with it.
+
+use std::path::PathBuf;
+use std::time::Duration;
+
+use clap::{Parser, Subcommand};
+use sn_node::config::{
+    self, IfName, ListenUri, MulticastMode, NodeSettings, PeerUri, PublicKeyHex,
+};
+use sn_node::yggdrasil::{self, AdminEndpoint};
+
+#[derive(Debug, Parser)]
+#[command(
+    name = "sn-node",
+    about = "SpiderNet node helper: Yggdrasil config generation + overlay status"
+)]
+struct Args {
+    #[command(subcommand)]
+    command: Command,
+}
+
+#[derive(Debug, Subcommand)]
+enum Command {
+    /// Generate a Yggdrasil config file for this SpiderNet node.
+    Genconf {
+        /// Peering URIs to connect to, comma separated (tcp://, tls://,
+        /// quic://, socks://).
+        #[arg(long = "peer", value_delimiter = ',')]
+        peers: Vec<String>,
+
+        /// Listener URIs for incoming peerings, comma separated
+        /// (e.g. `tcp://[::]:1337`).
+        #[arg(long = "listen", value_delimiter = ',')]
+        listen: Vec<String>,
+
+        /// Admin socket endpoint (tcp://host:port or unix:///path).
+        #[arg(long, default_value = "tcp://localhost:9001")]
+        admin: String,
+
+        /// Public keys of neighbors allowed to peer with us (64 hex
+        /// characters, comma separated). Empty: any node may peer.
+        #[arg(long = "allow-key", value_delimiter = ',')]
+        allow_key: Vec<String>,
+
+        /// Store the node's private key in an external file instead of
+        /// embedding it in the config.
+        #[arg(long)]
+        private_key_path: Option<PathBuf>,
+
+        /// TUN interface: `auto`, `none` (headless router-only), or a name.
+        #[arg(long, default_value = "auto")]
+        ifname: String,
+
+        /// Disable multicast (same-LAN) auto-peering.
+        #[arg(long)]
+        no_multicast: bool,
+
+        /// Path to the yggdrasil binary.
+        #[arg(long, default_value = "yggdrasil")]
+        yggdrasil_bin: String,
+
+        /// Output path for the generated config.
+        #[arg(short, long, default_value = "yggdrasil.conf")]
+        out: PathBuf,
+    },
+    /// Query the local Yggdrasil node via the admin socket.
+    Status {
+        /// Admin socket endpoint (tcp://host:port or unix:///path).
+        #[arg(long, default_value = "tcp://localhost:9001")]
+        admin: String,
+
+        /// Operation timeout in seconds.
+        #[arg(long, default_value_t = 3)]
+        timeout_secs: u64,
+    },
+}
+
+fn parse_ifname(input: &str) -> Result<IfName, String> {
+    match input.trim() {
+        "auto" => Ok(IfName::Auto),
+        "none" => Ok(IfName::Headless),
+        name if !name.is_empty() => Ok(IfName::Named(name.to_string())),
+        _ => Err("--ifname must be `auto`, `none` or an interface name".to_string()),
+    }
+}
+
+fn parse_uris<T>(
+    inputs: &[String],
+    what: &str,
+    parse: fn(&str) -> Result<T, sn_node::error::NodeError>,
+) -> Result<Vec<T>, String> {
+    inputs
+        .iter()
+        .map(|input| parse(input).map_err(|err| format!("{what}: {err}")))
+        .collect()
+}
+
+#[tokio::main]
+async fn main() {
+    if let Err(report) = run().await {
+        eprintln!("NO WARRANTY — SpiderNet sn-node (AGPL-3.0, no liability)");
+        eprintln!("error: {report}");
+        std::process::exit(1);
+    }
+}
+
+async fn run() -> Result<(), String> {
+    let args = Args::parse();
+    match args.command {
+        Command::Genconf {
+            peers,
+            listen,
+            admin,
+            allow_key,
+            private_key_path,
+            ifname,
+            no_multicast,
+            yggdrasil_bin,
+            out,
+        } => {
+            run_genconf(
+                peers,
+                listen,
+                admin,
+                allow_key,
+                private_key_path,
+                ifname,
+                no_multicast,
+                yggdrasil_bin,
+                out,
+            )
+            .await
+        }
+        Command::Status {
+            admin,
+            timeout_secs,
+        } => run_status(&admin, timeout_secs).await,
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn run_genconf(
+    peers: Vec<String>,
+    listen: Vec<String>,
+    admin: String,
+    allow_key: Vec<String>,
+    private_key_path: Option<PathBuf>,
+    ifname: String,
+    no_multicast: bool,
+    yggdrasil_bin: String,
+    out: PathBuf,
+) -> Result<(), String> {
+    let peers = parse_uris(&peers, "peer", PeerUri::parse)?;
+    let listen = parse_uris(&listen, "listen", ListenUri::parse)?;
+    let admin_listen = AdminEndpoint::parse(&admin).map_err(|err| err.to_string())?;
+    let allowed_public_keys = allow_key
+        .iter()
+        .map(|key| PublicKeyHex::parse(key).map_err(|err| err.to_string()))
+        .collect::<Result<Vec<_>, String>>()?;
+    let if_name = parse_ifname(&ifname)?;
+
+    let settings = NodeSettings {
+        peers,
+        listen,
+        multicast: Some(if no_multicast {
+            MulticastMode::Disabled
+        } else {
+            MulticastMode::PlatformDefault
+        }),
+        admin_listen: Some(admin_listen),
+        if_name: Some(if_name),
+        allowed_public_keys,
+        private_key_path,
+    };
+
+    let json = config::generate_config(&yggdrasil_bin, &settings).map_err(|err| err.to_string())?;
+    config::write_config(&out, &json).map_err(|err| err.to_string())?;
+
+    println!("NO WARRANTY — SpiderNet sn-node (AGPL-3.0, no liability)");
+    println!();
+    println!("config written: {}", out.display());
+    println!(
+        "mode: start with `yggdrasil -useconffile {}`",
+        out.display()
+    );
+    println!();
+    println!(
+        "NOTE: the config contains this node's PRIVATE key. Keep it at \
+         permission 600, never commit it, never share it."
+    );
+    Ok(())
+}
+
+async fn run_status(admin: &str, timeout_secs: u64) -> Result<(), String> {
+    let endpoint = AdminEndpoint::parse(admin).map_err(|err| err.to_string())?;
+    let timeout = Duration::from_secs(timeout_secs);
+
+    let info = yggdrasil::AdminClient::get_self(&endpoint, timeout)
+        .await
+        .map_err(|err| err.to_string())?;
+
+    println!("NO WARRANTY — SpiderNet sn-node (AGPL-3.0, no liability)");
+    println!();
+    println!("address:   {}", info.address);
+    if let Some(subnet) = info.subnet.as_ref() {
+        println!("subnet:    {subnet}");
+    }
+    if let Some(version) = info.build_version.as_ref() {
+        let name = info.build_name.as_deref().unwrap_or("yggdrasil");
+        println!("version:   {name} {version}");
+    }
+    println!("key:       {}", info.key);
+
+    let peers = yggdrasil::AdminClient::get_peers(&endpoint, timeout)
+        .await
+        .map_err(|err| err.to_string())?;
+    println!("peers:     {}", peers.len());
+    for peer in &peers {
+        match peer.remote.as_ref() {
+            Some(remote) => println!("  {} via {remote}", peer.address),
+            None => println!("  {}", peer.address),
+        }
+    }
+    Ok(())
+}
@@ -0,0 +1,268 @@
+//! Client for the Yggdrasil admin socket.
+//!
+//! Protocol (per the official docs): JSON stanza + `\n` over a TCP or unix
+//! socket; the default endpoint is `localhost:9001`. A response always has
+//! a `"status"` field (`"success"` or `"error"`), optionally a `"response"`
+//! section and an `"error"` field. By default Yggdrasil closes the
+//! connection after answering, so this client sends one request per
+//! connection.
+
+use std::net::Ipv6Addr;
+use std::path::PathBuf;
+use std::time::Duration;
+
+use crate::error::NodeError;
+
+/// Where the admin socket lives, in Yggdrasil endpoint syntax
+/// (`tcp://host:port` or `unix:///path/to/socket`).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum AdminEndpoint {
+    Tcp { host: String, port: u16 },
+    Unix { path: PathBuf },
+}
+
+impl AdminEndpoint {
+    /// Yggdrasil's default admin endpoint: `tcp://localhost:9001`.
+    pub fn default_local() -> Self {
+        Self::Tcp {
+            host: "localhost".to_string(),
+            port: 9001,
+        }
+    }
+
+    pub fn parse(input: &str) -> Result<Self, NodeError> {
+        let trimmed = input.trim();
+        if let Some(rest) = trimmed.strip_prefix("tcp://") {
+            let (host, port) = rest
+                .rsplit_once(':')
+                .ok_or_else(|| NodeError::InvalidAdminEndpoint(trimmed.to_string()))?;
+            let port: u16 = port
+                .parse()
+                .map_err(|_| NodeError::InvalidAdminEndpoint(trimmed.to_string()))?;
+            if host.is_empty() {
+                return Err(NodeError::InvalidAdminEndpoint(trimmed.to_string()));
+            }
+            Ok(Self::Tcp {
+                host: host.to_string(),
+                port,
+            })
+        } else if let Some(rest) = trimmed.strip_prefix("unix://") {
+            if rest.is_empty() {
+                return Err(NodeError::InvalidAdminEndpoint(trimmed.to_string()));
+            }
+            Ok(Self::Unix {
+                path: PathBuf::from(rest),
+            })
+        } else {
+            Err(NodeError::InvalidAdminEndpoint(trimmed.to_string()))
+        }
+    }
+
+    /// Value for the `AdminListen` field of a Yggdrasil config file.
+    pub fn config_value(&self) -> String {
+        match self {
+            Self::Tcp { host, port } => format!("tcp://{host}:{port}"),
+            Self::Unix { path } => format!("unix://{}", path.display()),
+        }
+    }
+}
+
+/// The node's own Yggdrasil address — a validated IPv6 address inside the
+/// Yggdrasil range.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct YggAddr(Ipv6Addr);
+
+impl YggAddr {
+    pub const fn new(addr: Ipv6Addr) -> Self {
+        Self(addr)
+    }
+
+    pub const fn inner(self) -> Ipv6Addr {
+        self.0
+    }
+}
+
+impl std::fmt::Display for YggAddr {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+/// One connection to the admin socket; one request per connection.
+pub struct AdminClient {
+    stream: AdminStream,
+}
+
+enum AdminStream {
+    Tcp(tokio::net::TcpStream),
+    Unix(tokio::net::UnixStream),
+}
+
+async fn connect_stream(endpoint: &AdminEndpoint) -> Result<AdminStream, NodeError> {
+    match endpoint {
+        AdminEndpoint::Tcp { host, port } => tokio::net::TcpStream::connect((host.as_str(), *port))
+            .await
+            .map(AdminStream::Tcp)
+            .map_err(NodeError::Io),
+        AdminEndpoint::Unix { path } => tokio::net::UnixStream::connect(path)
+            .await
+            .map(AdminStream::Unix)
+            .map_err(NodeError::Io),
+    }
+}
+
+impl AdminClient {
+    pub async fn connect(endpoint: &AdminEndpoint, timeout: Duration) -> Result<Self, NodeError> {
+        let stream = tokio::time::timeout(timeout, connect_stream(endpoint))
+            .await
+            .map_err(|_| NodeError::Timeout)??;
+        Ok(Self { stream })
+    }
+
+    /// Send one request verb (e.g. `getself`, `getpeers`) and return the
+    /// raw response section.
+    pub async fn request(&mut self, verb: &str) -> Result<serde_json::Value, NodeError> {
+        // `verb` is a compile-time-known constant at every call site, not
+        // user input — no injection risk.
+        let request = format!("{{\"request\":\"{verb}\"}}\n");
+        let bytes = request.as_bytes();
+
+        match &mut self.stream {
+            AdminStream::Tcp(stream) => {
+                tokio::io::AsyncWriteExt::write_all(stream, bytes).await?;
+                let mut buffer = Vec::new();
+                tokio::io::AsyncReadExt::read_to_end(stream, &mut buffer).await?;
+                Self::parse_response(&buffer, verb)
+            }
+            AdminStream::Unix(stream) => {
+                tokio::io::AsyncWriteExt::write_all(stream, bytes).await?;
+                let mut buffer = Vec::new();
+                tokio::io::AsyncReadExt::read_to_end(stream, &mut buffer).await?;
+                Self::parse_response(&buffer, verb)
+            }
+        }
+    }
+
+    /// Query the node's own status (address, subnet, key, build info).
+    pub async fn get_self(
+        endpoint: &AdminEndpoint,
+        timeout: Duration,
+    ) -> Result<SelfInfo, NodeError> {
+        let mut client = Self::connect(endpoint, timeout).await?;
+        let value = client.request("getself").await?;
+        SelfInfo::from_response(&value)
+    }
+
+    /// Query the node's active peer sessions.
+    pub async fn get_peers(
+        endpoint: &AdminEndpoint,
+        timeout: Duration,
+    ) -> Result<Vec<PeerInfo>, NodeError> {
+        let mut client = Self::connect(endpoint, timeout).await?;
+        let value = client.request("getpeers").await?;
+        peers_from_response(&value)
+    }
+
+    fn parse_response(buffer: &[u8], verb: &str) -> Result<serde_json::Value, NodeError> {
+        let envelope: serde_json::Value = serde_json::from_slice(buffer)?;
+        let status = envelope
+            .get("status")
+            .and_then(serde_json::Value::as_str)
+            .ok_or_else(|| {
+                NodeError::AdminProtocol(format!("{verb}: response has no `status` field"))
+            })?;
+        if status != "success" {
+            let error = envelope
+                .get("error")
+                .and_then(serde_json::Value::as_str)
+                .unwrap_or("unknown error");
+            return Err(NodeError::AdminFailed(error.to_string()));
+        }
+        envelope
+            .get("response")
+            .cloned()
+            .ok_or_else(|| NodeError::AdminProtocol(format!("{verb}: response has no data")))
+    }
+}
+
+/// Information about the local Yggdrasil node (`getSelf`).
+#[derive(Debug, Clone)]
+pub struct SelfInfo {
+    pub address: YggAddr,
+    pub key: String,
+    pub subnet: Option<String>,
+    pub coords: Option<serde_json::Value>,
+    pub build_name: Option<String>,
+    pub build_version: Option<String>,
+}
+
+impl SelfInfo {
+    /// Parse the `getSelf` response section: exactly one record, keyed by
+    /// the node's IPv6 address.
+    pub fn from_response(value: &serde_json::Value) -> Result<Self, NodeError> {
+        let obj = value.as_object().ok_or_else(|| {
+            NodeError::AdminProtocol("getSelf: response section is not an object".to_string())
+        })?;
+        let (address, record) = obj.iter().next().ok_or_else(|| {
+            NodeError::AdminProtocol("getSelf: response contains no record".to_string())
+        })?;
+        let address = YggAddr::new(address.parse::<Ipv6Addr>().map_err(|_| {
+            NodeError::AdminProtocol(format!("getSelf: `{address}` is not an IPv6 address"))
+        })?);
+
+        let str_field = |name: &str| {
+            record
+                .get(name)
+                .and_then(serde_json::Value::as_str)
+                .map(str::to_string)
+        };
+
+        Ok(Self {
+            address,
+            key: str_field("key").ok_or_else(|| {
+                NodeError::AdminProtocol("getSelf: record has no `key`".to_string())
+            })?,
+            subnet: str_field("subnet"),
+            coords: record.get("coords").cloned(),
+            build_name: str_field("build_name"),
+            build_version: str_field("build_version"),
+        })
+    }
+}
+
+/// One active peer session (`getPeers`).
+#[derive(Debug, Clone)]
+pub struct PeerInfo {
+    pub address: YggAddr,
+    pub key: Option<String>,
+    /// The peering URI of the remote peer, when reported.
+    pub remote: Option<String>,
+    pub coords: Option<serde_json::Value>,
+}
+
+/// Parse the `getPeers` response section: records keyed by IPv6 address.
+/// The first record typically refers to the current node itself.
+pub fn peers_from_response(value: &serde_json::Value) -> Result<Vec<PeerInfo>, NodeError> {
+    let obj = value.as_object().ok_or_else(|| {
+        NodeError::AdminProtocol("getPeers: response section is not an object".to_string())
+    })?;
+    let mut peers = Vec::with_capacity(obj.len());
+    for (address, record) in obj {
+        let address = YggAddr::new(address.parse::<Ipv6Addr>().map_err(|_| {
+            NodeError::AdminProtocol(format!("getPeers: `{address}` is not an IPv6 address"))
+        })?);
+        peers.push(PeerInfo {
+            address,
+            key: record
+                .get("key")
+                .and_then(serde_json::Value::as_str)
+                .map(str::to_string),
+            remote: record
+                .get("remote")
+                .and_then(serde_json::Value::as_str)
+                .map(str::to_string),
+            coords: record.get("coords").cloned(),
+        });
+    }
+    Ok(peers)
+}
@@ -0,0 +1,126 @@
+//! Integration tests for the Yggdrasil admin socket client, against a
+//! fake admin server speaking the documented protocol.
+
+#![forbid(unsafe_code)]
+#![allow(clippy::unwrap_used, clippy::expect_used)]
+
+use std::time::Duration;
+
+use sn_node::error::NodeError;
+use sn_node::yggdrasil::{peers_from_response, AdminClient, AdminEndpoint, SelfInfo};
+
+const SELF_RESPONSE: &str = concat!(
+    r#"{"request": {"request": "getself"}, "status": "success", "response": {"#,
+    r#""200:1111:2222:3333:4444:5555:6666:7777": {"#,
+    r#""address": "200:1111:2222:3333:4444:5555:6666:7777","#,
+    r#""key": "0000aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1111","#,
+    r#""build_name": "yggdrasil", "build_version": "0.5.12","#,
+    r#""subnet": "300:1111:2222:3333::/64", "coords": [1, 2, 3]}}}"#
+);
+
+const PEERS_RESPONSE: &str = concat!(
+    r#"{"request": {"request": "getpeers"}, "status": "success", "response": {"#,
+    r#""200:1111:2222:3333:4444:5555:6666:7777": {"coords": [1]}, "#,
+    r#""200:9999:2222:3333:4444:5555:6666:7777": {"#,
+    r#""coords": [1, 2], "port": 1, "key": "ffff...", "#,
+    r#""remote": "tcp://[fe80::1%25eth0]:1337"}}}"#
+);
+
+const ERROR_RESPONSE: &str =
+    r#"{"request": {"request": "getself"}, "status": "error", "error": "boom"}"#;
+
+/// Fake admin server: answers each connection's request line, then closes.
+async fn fake_admin_server(response: &'static str) -> std::net::SocketAddr {
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        loop {
+            let Ok((mut socket, _)) = listener.accept().await else {
+                break;
+            };
+            let mut line = Vec::new();
+            let mut byte = [0u8; 1];
+            while !line.ends_with(b"\n") {
+                if socket.read(&mut byte).await.unwrap_or(0) == 0 {
+                    break;
+                }
+                line.push(byte[0]);
+            }
+            socket.write_all(response.as_bytes()).await.ok();
+            socket.shutdown().await.ok();
+        }
+    });
+    addr
+}
+
+fn tcp_endpoint(addr: std::net::SocketAddr) -> AdminEndpoint {
+    AdminEndpoint::Tcp {
+        host: addr.ip().to_string(),
+        port: addr.port(),
+    }
+}
+
+#[tokio::test]
+async fn get_self_parses_address_subnet_and_version() {
+    let endpoint = tcp_endpoint(fake_admin_server(SELF_RESPONSE).await);
+    let info = SelfInfo::from_response(
+        &AdminClient::connect(&endpoint, Duration::from_secs(2))
+            .await
+            .unwrap()
+            .request("getself")
+            .await
+            .unwrap(),
+    )
+    .unwrap();
+
+    assert_eq!(
+        info.address.inner().to_string(),
+        "200:1111:2222:3333:4444:5555:6666:7777"
+    );
+    assert_eq!(info.subnet.as_deref(), Some("300:1111:2222:3333::/64"));
+    assert_eq!(info.build_version.as_deref(), Some("0.5.12"));
+}
+
+#[tokio::test]
+async fn get_peers_lists_remote_uris() {
+    let endpoint = tcp_endpoint(fake_admin_server(PEERS_RESPONSE).await);
+    let value = AdminClient::connect(&endpoint, Duration::from_secs(2))
+        .await
+        .unwrap()
+        .request("getpeers")
+        .await
+        .unwrap();
+    let peers = peers_from_response(&value).unwrap();
+
+    assert_eq!(peers.len(), 2);
+    assert_eq!(
+        peers[1].remote.as_deref(),
+        Some("tcp://[fe80::1%25eth0]:1337")
+    );
+}
+
+#[tokio::test]
+async fn admin_error_is_surfaced() {
+    let endpoint = tcp_endpoint(fake_admin_server(ERROR_RESPONSE).await);
+    let mut client = AdminClient::connect(&endpoint, Duration::from_secs(2))
+        .await
+        .unwrap();
+    let err = client.request("getself").await.unwrap_err();
+    assert!(matches!(err, NodeError::AdminFailed(ref text) if text == "boom"));
+}
+
+#[tokio::test]
+async fn connection_refused_becomes_io_or_timeout_error() {
+    // Port 1 on localhost: virtually guaranteed to be closed.
+    let endpoint = AdminEndpoint::Tcp {
+        host: "127.0.0.1".into(),
+        port: 1,
+    };
+    let result = AdminClient::get_self(&endpoint, Duration::from_millis(500)).await;
+    assert!(matches!(
+        result,
+        Err(NodeError::Io(_)) | Err(NodeError::Timeout)
+    ));
+}
Author	SHA1	Message	Date
Ben_Kosytorz	50e459ed79	Docs: Yggdrasil overlay note (rationale, deployment, admin API) + link updates	2026-09-24 16:00:43 +02:00
Ben_Kosytorz	c422deee7c	sn-node: Yggdrasil config generation and admin socket client (genconf, status)	2026-09-24 15:58:12 +02:00
Ben_Kosytorz	23dc906da0	sn-fetch: per-exit egress (Direct vs Proxy) for neighbor exits over the mesh	2026-09-24 15:58:12 +02:00
Ben_Kosytorz	810529d502	sn-fetch: per-segment retry with exit reassignment and progress sinks	2026-09-24 15:58:11 +02:00