Add pilot handbook and document sn-fair v0

pilot.md: step-by-step for connecting two households (link-agnostic), measurement protocol M0-M4 (link baseline, uplink baseline, pooled, concurrent downloads, sn-fair Soll/Ist comparison), safety/legal checklist, and a results section to be filled after the pilot. Cross-linked from Home, documentation README, roadmap, architecture and server README; roadmap marks sn-fair v0 done.
Add sn-fair v0: weighted max-min fairness scheduler with leech policy
2026-09-24 18:46:47 +02:00 · 2026-09-24 18:46:38 +02:00
15 changed files with 1279 additions and 14 deletions
@@ -25,7 +25,7 @@ Datensammlung, keine Bezahlung.
 | Ordner/Datei | Inhalt | Einstieg |
 |---|---|---|
 | `documentation/` | Idee, Ethos, Architektur, Rechtslage, Roadmap (Deutsch, Obsidian-freundlich mit `[[Wikilinks]]`; `documentation/Home.md` ist die Map of Content) | [README](documentation/README.md) |
-| `server/` | Die Software (Rust): `sn-fetch`, `sn-node`, `sn-exit`; geplant `sn-fair`, `sn-cache`, `sn-freenet` | [README](server/README.md) |
+| `server/` | Die Software (Rust): `sn-fetch`, `sn-node`, `sn-exit`, `sn-fair`; geplant `sn-cache`, `sn-freenet` | [README](server/README.md) |
 | `scripts/` | Repo-Setup (`bootstrap-gitea.sh`) | — |
 | `AGENTS.md` | Arbeitsanweisungen für Coding-Agents und Contributor:innen | [AGENTS.md](AGENTS.md) |
 | `LICENSE` | Code-Lizenz: AGPL-3.0-only | [LICENSE](LICENSE) |
@@ -53,8 +53,11 @@ Prototyp (Konzeptphase + erster Code):
 - **`sn-exit` v0** funktioniert: kontingentierter Uplink-Proxy (Byte-Relay,
  CONNECT-Tunnel für HTTPS, SSRF-Schutz, Source-Allowlist) — End-to-End
  mit `sn-fetch` gegen einen echten Server verifiziert
+- **`sn-fair` v0** funktioniert: Fairness-Engine (gewichtetes Max-Min,
+  Leech-Politik) + Demo-CLI — die Soll-Aufteilung für den Piloten
 - **Nächste Schritte:** Pilot mit 2–3 Haushalten (Yggdrasil-Overlay
-  verkabeln, Messwerte), Fairness-Scheduler — siehe
+  verkabeln, Messwerte — Handbuch:
+  [pilot.md](documentation/ideen/pilot.md)), `sn-node`-Daemon — siehe
  [Roadmap](documentation/ideen/roadmap.md)

 ## Doku lesen (Obsidian)
@@ -26,6 +26,7 @@ Nachbarschaft betreibt ihre eigene Instanz nach eigenen Regeln.
 | [[Splitting]] | Was sich splitten lässt: Ranges, Flow-Verteilung, Realtime |
 | [[Lizenz]] | Lizenz-Optionen (Code & Doku) |
 | [[Roadmap]] | Phasen: Pilot → MVP → Rollout → Freenet → Backbone |
+| [[Pilot]] | Pilot-Handbuch: 2 Haushalte verkabeln, messen, dokumentieren |
 | [[Server-README|Server-Konzept]] | Software-Komponenten, Code-Standards |

 ## Lesereihenfolge zum Einstieg
@@ -44,6 +45,8 @@ Nachbarschaft betreibt ihre eigene Instanz nach eigenen Regeln.
 - [[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`
+- [[Pilot]] ↔ [[Roadmap]] (Phase 1) und [[Overlay]] §Deployment — der Pilot ist der erste Real-Test
+- [[Pilot]] §M4 ↔ [[Architektur]] §Fairness-Scheduler — `sn-fair` berechnet die Soll-Aufteilung
 - [[Lizenz]] ↔ [[Ethos]] §6 (Transparenz) und `freenet-core` (AGPL)

 ## Außerhalb der Doku
@@ -17,6 +17,7 @@ unabhängige SpiderNet-Instanzen geben; wir providen nur die Werkzeuge.
 - [[Splitting]] — Was sich über mehrere Exits splitten/verteilen lässt (und was nicht)
 - [[Lizenz]] — Lizenz-Optionen und Empfehlung
 - [[Roadmap]] — Phasenplan vom Pilot bis zum Nachbarschafts-Backbone
+- [[Pilot]] — Pilot-Handbuch: Schritt-für-Schritt + Messprotokoll (Phase 1)
 - [[Server-README|Server-Konzept]] — geplante Software (`sn-node`, `sn-fetch`, …)

 ## Die Idee in einem Satz
@@ -29,10 +30,11 @@ ist — ohne Mehrkosten und mit besserer Versorgung schwacher Haushalte.
 ## Status

 Prototypphase (siehe [[Roadmap]]): `sn-fetch` (segmentierter Multi-Exit-
-Download mit Retry + Egress), `sn-node` v0 (Yggdrasil-Config + Status)
-und `sn-exit` v0 (kontingentierter Uplink-Proxy mit SSRF-Schutz) sind
-implementiert und getestet — Details in [`server/README.md`](../server/README.md).
-Nächster Meilenstein: Pilot mit 2–3 Haushalten (echtes Overlay, Messwerte).
+Download mit Retry + Egress), `sn-node` v0 (Yggdrasil-Config + Status),
+`sn-exit` v0 (kontingentierter Uplink-Proxy mit SSRF-Schutz) und `sn-fair`
+v0 (Fairness-Engine + Demo-CLI) sind implementiert und getestet —
+Details in [`server/README.md`](../server/README.md). Nächster
+Meilenstein: Pilot mit 2–3 Haushalten (Handbuch: [[Pilot]]).

 ## Verwandte Projekte (Prior Art)

@@ -133,6 +133,11 @@ siehe [[Splitting]].
 - Exit-Kontingente (Volumenlimit, Wartungszeit) werden respektiert
 - Leech-Nodes: niedrigste Priorität bei Knappheit, nie ausgeschlossen

+Status: **v0 implementiert** — gewichtetes Max-Min in zwei Klassen:
+Contributors teilen die volle Kapazität nach Max-Min, Leeches teilen
+die Restkapazität (bei Sättigung zuerst gedrosselt, nie ausgeschlossen).
+Details in `server/README.md`; Soll-Ist-Vergleich im Piloten: [[Pilot]] §M4.
+
 ### Nachbarschafts-Cache (`sn-cache`)

 - Transparenter Shared-Cache (lancache-Prinzip): große, populäre Objekte
@@ -0,0 +1,206 @@
+# Pilot: zwei Haushalte verbinden und messen
+
+> Prototyp-Betrieb, **keine Garantie** (AGPL §§ 15/16) — Nutzung auf eigene
+> Verantwortung, siehe [[Recht]] und [[Server-README]]. Persönliche Daten
+> (Namen, Adressen, Keys realer Nachbarn) bleiben **außerhalb des Repos**.
+
+Ziel der Phase 1 (siehe [[Roadmap]]) ist der Beweis, dass Bandbreiten-
+Pooling über zwei echte Anschlüsse real funktioniert — nicht auf dem
+Papier. Der Pilot liefert:
+
+1. **Messwerte:** 1-Exit-Baseline vs. 2-Exit-gepoolt (gleiche Datei)
+2. **Fairness-Daten:** zwei gleichzeitige Downloads teilen sich den Pool
+3. **Stolpersteine:** was die Doku/Software dadurch lernen muss
+
+Der Pilot ist **link-agnostisch**: Ob die Häuser per Glasfaser,
+SFP+-Direktkabel, Switch im Keller, 60-GHz-Richtfunk oder normalem
+Ethernet verbunden sind, entscheidet die Nachbarschaft — die Software
+misst, was da ist (siehe [[Architektur]], Schicht 0). Alle
+Angaben zu Link-Typen in dieser Datei sind Beispiele.
+
+## Voraussetzungen
+
+### Teilnehmer
+
+- 2 Haushalte (ideal 3, der dritte als **Leech** — Empfang ohne
+  Exit-Angebot, um genau diesen Modus mitzutesten)
+- Ein persönlicher Termin; technisches Onboarding gesicht zu gesicht
+  (Nachbarn sind der Trust-Anchor, siehe [[Architektur]])
+
+### Hardware (je Haushalt, Beispiele)
+
+- Ein Mini-PC oder Einplatinenrechner (≥ 2 GB RAM, Debian o. ä.) als
+  Node-Host — je eine NIC Richtung Heim-Router und Richtung
+  Nachbarschafts-Link (übergangsweise reicht auch eine NIC am
+  Mesh-Switch)
+- Verbindungsweg — **eins davon genügt:**
+  - Glasfaser/DAC-Direktkabel zwischen den SFP/SFP+-Ports
+  - gemeinsamer Switch im Keller/Treppenhaus
+  - 60-GHz-Richtfunk-Pair (Sichtlinie nötig, siehe [[Recht]] §3)
+  - vorhandenes Ethernet/WLAN (langsamer, aber null Aufwand)
+
+### Software
+
+- [yggdrasil](https://yggdrasil-network.github.io/) (Binary oder
+  Debian-Paket)
+- SpiderNet-Binaries bauen und rüberkopieren:
+
+```sh
+cd server
+cargo build --release --workspace
+# → target/release/sn-node, sn-exit, sn-fetch, sn-fair
+```
+
+### Vor dem Termin (Checkliste)
+
+- [ ] ISP-AGB der Teilnehmenden geprüft (Weitergabe an Dritte,
+  siehe [[Recht]] §2)
+- [ ] Nachbarschafts-Abkommen geklärt: wer spendiert wie viel
+  (Quota), Leechen ausdrücklich erlaubt, alles jederzeit widerruflich
+- [ ] Hardware vorhanden und bootfähig
+- [ ] Binaries gebaut und auf beide Nodes kopiert
+- [ ] Testdatei gewählt: Range-fähiger Server, ~750 MB, bekannte
+  SHA-256 (im PoC hat sich eine Debian-Netinst-ISO bewährt; die
+  offizielle `SHA256SUMS` der jeweiligen Version nutzen)
+
+## Termintag: Schritt-für-Schritt
+
+Die technischen Details (Config, Schlüssel, Admin-Socket) stehen in
+[[Overlay]] §Deployment — hier die Reihenfolge mit Checkpunkten.
+
+1. **Physik verbinden.** Link herstellen, L2-Test: `ping` über die
+   lokale Verbindung. Optional `iperf3` über den Link → das ist
+   **Messpunkt M0** (Kapazität des Nachbarschafts-Links).
+2. **Yggdrasil einrichten (beide Häuser).**
+
+   ```sh
+   sn-node genconf --listen "tcp://[::]:1337" \
+     --peer "tcp://[<Mesh-IP des Nachbarn>]:1337" \
+     --allow-key <Public-Key des Nachbarn> \
+     --private-key-path /etc/spidernet/node.key \
+     --out /etc/yggdrasil.conf
+   yggdrasil -useconffile /etc/yggdrasil.conf
+   sn-node status   # eigene Overlay-Adresse + Peering-Status
+   ```
+
+   - Im selben LAN reicht Multicast-Auto-Peering (ohne `--peer`).
+   - Config enthält den **privaten** Key (`chmod 600` macht
+     `genconf` selbst) — niemals teilen oder committen.
+   - Die eigenen Adressen/Keys werden **persönlich** ausgetauscht,
+     nicht per Chat/Repo.
+3. **Overlay-Test.** `sn-node status` zeigt auf *beiden* Seiten den
+   Nachbarn als Peer; `ping <Overlay-Adresse des Nachbarn>` geht.
+4. **Exit starten (Haus B spendiert):**
+
+   ```sh
+   sn-exit --bind "[<Overlay-Adresse B>]:8080" \
+     --allow <Overlay-Adresse A> \
+     --quota-mib 1024
+   ```
+
+   - `--allow` nie offen lassen (ohne Liste ist nur Loopback
+     erlaubt, das ist der Dev-Mode).
+   - Test von Haus A: ein HTTP-Request gegen einen Origin über
+     `http://[<Overlay-Adresse B>]:8080` als Proxy (z. B. mit
+     `sn-fetch --probe-only` oder `curl -x`).
+5. **Erster gepoolter Download (Haus A):**
+
+   ```sh
+   sn-fetch <url> -o testdatei.iso \
+     -e "1=1,2=1@http://[<Overlay-Adresse B>]:8080" \
+     --sha256 <erwarteter-hash>
+   ```
+
+6. **Kill-Switch-Test.** `sn-exit` auf B stoppen → laufender und
+   nächster Download fällt auf den eigenen Uplink zurück (Retry mit
+   Exit-Reassignment). Danach B wieder starten. Beides dokumentieren:
+   Dauer der Störung, verlorene Segmente.
+
+Wichtig zur Beruhigung: SpiderNet v0 installiert **keine
+Routing-Änderung** im Heimnetz. Geräte nutzen wie gewohnt den eigenen
+Router; nur `sn-fetch`-Downloads gehen (explizit) in den Pool. Alles
+stoppen (`sn-exit`, `yggdrasil`) = das Haus ist wie vorher.
+
+## Messprotokoll
+
+Regeln für belastbare Werte:
+
+- Gleiche Testdatei, gleiche Uhrzeit ± 1 h (ISP-Congestion schwankt
+  tageszeitabhängig)
+- Pro Szenario **3 Läufe**, Median notieren; zwischen Läufen Pause
+- Nichts anderes auf den Nodes laufen lassen; Messwerte immer mit
+  Datum/Uhrzeit und den beidseitigen Uplink-Tarifen notieren
+
+### Szenarien
+
+| # | Szenario | Wo | Was gemessen wird |
+|---|---|---|---|
+| M0 | Nachbarschafts-Link-Kapazität | A↔B | `iperf3` über den lokalen Link |
+| M1 | Uplink-Baseline | je Haus | nur eigener Uplink (`-e "1=1"`), ISP-Kapazität |
+| M2 | Gepoolt, 1 Download | A | 2 Exits (`-e "1=1,2=1@http://[B]:8080"`), Gesamt-Durchsatz vs. M1 |
+| M3 | Gepoolt, Konkurrenz | A + B | zwei `sn-fetch`-Läufe gleichzeitig, beide über beide Exits — wie teilen sich die Aufstellung |
+| M4 | Fairness-Soll | (Simulation) | `sn-fair` mit den M1/M2-Werten → Soll-Aufteilung zum Vergleich mit M3 |
+
+### Tabelle M2/M3 (Vorlage, pro Lauf eine Zeile)
+
+| Lauf | Szenario | Datum/Zeit | Durchsatz Mbit/s | Dauer | Segmente | Retries | Bemerkung |
+|---|---|---|---|---|---|---|---|
+| 1 | M2 | | | | | | |
+| 2 | M2 | | | | | | |
+| 3 | M2 | | | | | | |
+
+Zusätzlich notieren: Quota-Rest auf B, CPU/RAM der Node-Hosts,
+(evtl.) Firmware/Treiber des Link-Wegs.
+
+### Fairness-Soll mit `sn-fair` (M4)
+
+Die gemessenen Uplink-Kapazitäten (M1) als Pool-Kapazität und die
+gleichzeitigen Nachfragen (M3) einsetzen — das CLI arbeitet in
+Mbit/s:
+
+```sh
+sn-fair --capacity <M1-A + M1-B, Mbit/s> \
+  --participant 1=contributing \
+  --participant 2=contributing \
+  --participant 3=leeching \
+  --demand 1=<Nachfrage A> \
+  --demand 2=<Nachfrage B> \
+  --demand 3=<Nachfrage Leech>
+```
+
+Erwartung (Regelwerk in [[Architektur]] §Fairness-Scheduler):
+Contributors bekommen die Kapazität nach max-min Fairness (gewichtet),
+Leeches profitieren von Restkapazität — bei Sättigung zuerst gedrosselt,
+nie ausgeschlossen. Weicht M3 (Ist) deutlich von M4 (Soll) ab, landet
+das als Erkenntnis in der Ergebnisse-Sektion unten (die echte
+Runtime-Integration in `sn-fetch` ist Phase 2, siehe [[Roadmap]]).
+
+## Sicherheit & Recht (Checkliste während des Piloten)
+
+- [ ] `sn-exit` nur mit `--allow` (Allowlist) und Quota laufen lassen
+- [ ] SSRF-Schutz nie deaktivieren (verhindert Nachbarn-Zugriff auf
+  Heim-LAN/Admin-Sockets des Exit-Haushalts)
+- [ ] Keine Port-Forwards ins Internet, außer ein Peering braucht es
+  zwingend (siehe [[Overlay]] §Grenzen)
+- [ ] Logs minimal und lokal (keine Inhaltslogs, siehe [[Recht]] §4)
+- [ ] Haftungsausschluss ist den Teilnehmenden bekannt gezeigt
+- [ ] Sperrbarkeit praktisch geübt: Exit stoppen = Kill-Switch,
+  Teilnehmersperre = `--allow`-Liste ändern
+
+## Ergebnisse
+
+*(wird nach dem Piloten gefüllt — Messwerte, Erkenntnisse und
+Stolpersteine gehören hier ins Repo, nicht in den Chat)*
+
+### Messwerte
+
+*(Tabelle ergänzen — M0 bis M4, Mediane der 3 Läufe)*
+
+### Erkenntnisse
+
+*(kurz, ehrlich, mit Quellen/Logs wo sinnvoll)*
+
+### Stolpersteine & Anpassungen
+
+*(was in [[Architektur]], [[Overlay]], [[Roadmap]] oder der Software
+geändert werden muss, damit der nächste Pilot weniger stolpert)*
@@ -14,6 +14,7 @@ endet mit etwas, das man benutzen kann.
 ## Phase 1 — Pilot (2–3 Haushalte)

 Ziel: Beweis, dass Pooling über zwei Anschlüsse real funktioniert.
+Handbuch mit Messprotokoll: [[Pilot]].

 - [x] `sn-fetch` PoC (lokal): Range-Probe, gewichteter Planner, parallele
      Segment-Downloads, Reassembly + SHA-256 — verifiziert gegen einen
@@ -27,10 +28,12 @@ Ziel: Beweis, dass Pooling über zwei Anschlüsse real funktioniert.
      WLAN/60-GHz), `sn-node`-Daemon orchestriert den Node
 - [x] `sn-exit` v0: HTTP-Proxy auf der Yggdrasil-Adresse (Kontingente,
      SSRF-Schutz, Kill-Switch = Prozess stoppen)
+- [x] `sn-fair` v0: Fairness-Engine (gewichtetes Max-Min über zwei
+      Klassen, Leech-Politik) + Demo-CLI; Runtime-Integration in
+      `sn-fetch` ist Phase 2
 - [ ] `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
- [ ] Doku: Messwerte, Erkenntnisse, Stolpersteine → hier in das Repo
+      (1-Exit vs. 2-Exit, gleiche Datei — Protokoll in [[Pilot]])
+- [ ] Doku: Messwerte, Erkenntnisse, Stolpersteine → in [[Pilot]] §Ergebnisse

 ## Phase 2 — MVP-Software

@@ -1009,6 +1009,25 @@ version = "1.16.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ba467056f1b547ed52077911161fc86985becbc60e8e1857c8a144dab0def891"

+[[package]]
+name = "sn-exit"
+version = "0.1.0"
+dependencies = [
+ "clap",
+ "thiserror",
+ "tokio",
+ "tracing",
+ "tracing-subscriber",
+]
+
+[[package]]
+name = "sn-fair"
+version = "0.1.0"
+dependencies = [
+ "clap",
+ "thiserror",
+]
+
 [[package]]
 name = "sn-fetch"
 version = "0.1.0"
@@ -1024,6 +1043,16 @@ dependencies = [
 "tracing-subscriber",
 ]

+[[package]]
+name = "sn-node"
+version = "0.1.0"
+dependencies = [
+ "clap",
+ "serde_json",
+ "thiserror",
+ "tokio",
+]
+
 [[package]]
 name = "socket2"
 version = "0.6.5"
@@ -1,6 +1,6 @@
 [workspace]
 resolver = "2"
-members = ["crates/sn-fetch", "crates/sn-node", "crates/sn-exit"]
+members = ["crates/sn-fetch", "crates/sn-node", "crates/sn-exit", "crates/sn-fair"]

 [workspace.package]
 version = "0.1.0"
@@ -60,10 +60,26 @@ SHA-256 entspricht der offiziellen `SHA256SUMS`).
 - Verifiziert end-to-end: `sn-fetch` über einen laufenden `sn-exit`
  lädt die Debian-ISO (756 MB, 4 Segmente, SHA-256 korrekt)

-**Noch nicht drin:** `sn-node`-Daemon (Supervisor, der yggdrasil + sn-exit
-+ sn-fetch orchestriert), Fairness-Scheduler.
+**`sn-fair` v0 ist implementiert** (siehe `crates/sn-fair/`):

-Details zum Overlay: [[Overlay]].
+- Gewichtetes Max-Min-Fairness-Engine (zwei Klassen): **Contributors**
+  (Haushalte mit Exit-Angebot, konfigurierbares Gewicht) teilen sich
+  die Gesamtkapazität nach Max-Min-Fairness; **Leeches** (Empfang ohne
+  Beitrag, ausdrücklich legitim) bekommen die Restkapazität — bei
+  Sättigung zuerst gedrosselt, nie ausgeschlossen
+- Deterministische, overflow-sichere Berechnung (u128-Arithmetik,
+  Restverteilung in ID-Ordnung), validierte Newtypes
+  (`ParticipantId`, `Weight`, `Rate`, `Capacity`)
+- Demo-CLI in Mbit/s für den Piloten: Soll-Aufteilung simulieren und
+  gegen die gemessene Ist-Aufteilung halten (siehe [[Pilot]], M4)
+- Runtime-Integration in `sn-fetch` (echte gleichzeitige Downloads
+  drosseln) ist Phase 2, siehe [[Roadmap]]
+
+**Noch nicht drin:** `sn-node`-Daemon (Supervisor, der yggdrasil +
+sn-exit + sn-fetch orchestriert), Freenet-Integration, Cache.
+
+Details zum Overlay: [[Overlay]]. Pilot-Handbuch mit Messprotokoll:
+[[Pilot]].

 ## Haftungsausschluss (muss im Produkt sichtbar sein)

@@ -96,7 +112,7 @@ Einplatinenrechner), der:
 | `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 (v0: Byte-Relay + Quota + SSRF-Schutz) |
 | `sn-fetch` | Segmentierter Multi-Exit-Downloader (HTTP-API + CLI) |
-| `sn-fair` | Fairness-Scheduler (max-min, Exit-Kontingente, Karma, Leech-Modus) |
+| `sn-fair` | Fairness-Scheduler (max-min, Exit-Kontingente, Karma, Leech-Modus) — v0: zwei-Klassen-Max-Min-Engine + Demo-CLI |
 | `sn-cache` | Transparenter Nachbarschafts-Cache (DNS-basiert) |
 | `sn-freenet` | Integration von `freenet-core` (Contracts für Nachbarschafts-Inhalte) |

@@ -0,0 +1,13 @@
+[package]
+name = "sn-fair"
+description = "Weighted max-min fairness scheduler for SpiderNet neighborhood networks"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+thiserror = { workspace = true }
+clap = { workspace = true }
+
+[lints]
+workspace = true
@@ -0,0 +1,24 @@
+//! Error type for the fairness scheduler.
+
+use crate::types::ParticipantId;
+
+#[derive(Debug, thiserror::Error)]
+pub enum FairnessError {
+    #[error("capacity must be greater than zero")]
+    ZeroCapacity,
+
+    #[error("invalid participant id `{0}`: expected a decimal number greater than zero")]
+    InvalidParticipantId(String),
+
+    #[error("demand references unknown participant {0}")]
+    UnknownParticipant(ParticipantId),
+
+    #[error("duplicate demand for participant {0}")]
+    DuplicateDemand(ParticipantId),
+
+    #[error("duplicate participant {0}")]
+    DuplicateParticipant(ParticipantId),
+
+    #[error("conflicting participation classes for participant {0}")]
+    ConflictingParticipation(ParticipantId),
+}
@@ -0,0 +1,578 @@
+//! Two-class weighted max-min fairness allocation.
+//!
+//! Participants are split into two priority classes:
+//!
+//! 1. **Contributors** ([`Participation::Contributing`]) — households that
+//!    offer their own uplink to the pool. They are served first, with
+//!    shares shaped by their [`Weight`] via weighted max-min fairness
+//!    over the *full* capacity.
+//! 2. **Leeches** ([`Participation::Leeching`]) — receive-only
+//!    households. Leech mode is explicitly allowed (the network exists
+//!    under neighbors, not against them). Leeches have the lowest but
+//!    still fair priority: when contributors saturate the capacity,
+//!    leeches get (near) nothing; when contributors are undersubscribed,
+//!    leeches benefit from the leftovers. Nobody is dropped from the
+//!    system.
+//!
+//! See the crate root for the phase overview.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use crate::error::FairnessError;
+use crate::types::{Capacity, ParticipantId, Rate, Weight};
+
+/// How a household takes part in the neighborhood network.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Participation {
+    /// The household offers its own uplink to the pool; `weight` shapes
+    /// its share of the capacity (1 = equal share).
+    Contributing { weight: Weight },
+    /// The household only receives (leech mode, explicitly allowed).
+    /// Leeches share whatever contributors leave unused, with implicit
+    /// equal weights, capped by their demand.
+    Leeching,
+}
+
+/// A participant of the allocation: an id plus its participation class.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Participant {
+    id: ParticipantId,
+    participation: Participation,
+}
+
+impl Participant {
+    pub const fn new(id: ParticipantId, participation: Participation) -> Self {
+        Self { id, participation }
+    }
+
+    pub const fn id(&self) -> ParticipantId {
+        self.id
+    }
+
+    pub const fn participation(&self) -> Participation {
+        self.participation
+    }
+}
+
+/// Requested rate of a participant.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Demand {
+    participant: ParticipantId,
+    rate: Rate,
+}
+
+impl Demand {
+    pub const fn new(participant: ParticipantId, rate: Rate) -> Self {
+        Self { participant, rate }
+    }
+
+    pub const fn participant(&self) -> ParticipantId {
+        self.participant
+    }
+
+    pub const fn rate(&self) -> Rate {
+        self.rate
+    }
+}
+
+/// Granted rate for a participant.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Allocation {
+    participant: ParticipantId,
+    rate: Rate,
+}
+
+impl Allocation {
+    pub const fn new(participant: ParticipantId, rate: Rate) -> Self {
+        Self { participant, rate }
+    }
+
+    pub const fn participant(&self) -> ParticipantId {
+        self.participant
+    }
+
+    pub const fn rate(&self) -> Rate {
+        self.rate
+    }
+}
+
+/// Result of [`allocate`]: per-participant rates plus the totals.
+///
+/// Allocations are keyed by participant id in a `BTreeMap`, so iteration
+/// is always in ascending id order (deterministic output).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct FairShare {
+    allocations: BTreeMap<ParticipantId, Rate>,
+    total_allocated: Rate,
+    unused: Rate,
+}
+
+impl FairShare {
+    /// Allocations ordered by participant id. Participants without a
+    /// demand at all are absent; use [`FairShare::rate_of`] to read a
+    /// single participant's rate.
+    pub fn allocations(&self) -> &BTreeMap<ParticipantId, Rate> {
+        &self.allocations
+    }
+
+    /// Allocations as [`Allocation`] values, in participant-id order.
+    pub fn iter(&self) -> impl Iterator<Item = Allocation> + '_ {
+        self.allocations
+            .iter()
+            .map(|(&participant, &rate)| Allocation::new(participant, rate))
+    }
+
+    /// Allocated rate of one participant (zero if it received nothing).
+    pub fn rate_of(&self, participant: ParticipantId) -> Rate {
+        self.allocations
+            .get(&participant)
+            .copied()
+            .unwrap_or(Rate::new(0))
+    }
+
+    pub const fn total_allocated(&self) -> Rate {
+        self.total_allocated
+    }
+
+    pub const fn unused(&self) -> Rate {
+        self.unused
+    }
+}
+
+/// Allocates `capacity` among `participants` according to their `demands`.
+///
+/// Phase 1 runs weighted max-min fairness among contributors over the
+/// full capacity; phase 2 runs weighted max-min among leeches (implicit
+/// equal weights) over the capacity the contributors leave unused.
+///
+/// Empty `demands` is valid: nothing is allocated, everything stays
+/// unused. Participants without a demand (or with demand 0) receive
+/// nothing.
+///
+/// Errors:
+/// - [`FairnessError::ZeroCapacity`] — capacity is zero (cannot happen
+///   with a validated [`Capacity`])
+/// - [`FairnessError::UnknownParticipant`] — a demand references an id
+///   that is not among `participants`
+/// - [`FairnessError::DuplicateDemand`] — two demands for the same id
+/// - [`FairnessError::DuplicateParticipant`] — the same id appears twice
+///   among `participants` with the same participation class
+/// - [`FairnessError::ConflictingParticipation`] — the same id appears
+///   twice among `participants` with different participation classes
+pub fn allocate(
+    capacity: Capacity,
+    participants: &[Participant],
+    demands: &[Demand],
+) -> Result<FairShare, FairnessError> {
+    let mut participation_of: BTreeMap<ParticipantId, Participation> = BTreeMap::new();
+    for participant in participants {
+        if let Some(previous) =
+            participation_of.insert(participant.id(), participant.participation())
+        {
+            if previous == participant.participation() {
+                return Err(FairnessError::DuplicateParticipant(participant.id()));
+            }
+            return Err(FairnessError::ConflictingParticipation(participant.id()));
+        }
+    }
+
+    for demand in demands {
+        if !participation_of.contains_key(&demand.participant()) {
+            return Err(FairnessError::UnknownParticipant(demand.participant()));
+        }
+    }
+
+    let mut demand_of: BTreeMap<ParticipantId, u64> = BTreeMap::new();
+    for demand in demands {
+        if demand_of
+            .insert(demand.participant(), demand.rate().value())
+            .is_some()
+        {
+            return Err(FairnessError::DuplicateDemand(demand.participant()));
+        }
+    }
+
+    let mut contributors = Vec::new();
+    let mut leeches = Vec::new();
+    for (&id, &participation) in &participation_of {
+        let demand = demand_of.get(&id).copied().unwrap_or(0);
+        match participation {
+            Participation::Contributing { weight } => contributors.push(Entry {
+                id,
+                demand,
+                weight: weight.value(),
+            }),
+            Participation::Leeching => leeches.push(Entry {
+                id,
+                demand,
+                weight: 1,
+            }),
+        }
+    }
+
+    let phase_one = weighted_max_min(contributors, capacity.value());
+    let allocated: u128 = phase_one.values().map(|rate| u128::from(*rate)).sum();
+    let leftover = capacity.value().saturating_sub(to_u64(allocated));
+
+    let phase_two = weighted_max_min(leeches, leftover);
+
+    let mut allocations = phase_one;
+    allocations.extend(phase_two);
+
+    let total: u128 = allocations.values().map(|rate| u128::from(*rate)).sum();
+    let total = to_u64(total);
+    let unused = capacity.value().saturating_sub(total);
+
+    Ok(FairShare {
+        allocations: allocations
+            .into_iter()
+            .map(|(id, rate)| (id, Rate::new(rate)))
+            .collect(),
+        total_allocated: Rate::new(total),
+        unused: Rate::new(unused),
+    })
+}
+
+/// Internal per-participant input for the max-min helper.
+struct Entry {
+    id: ParticipantId,
+    demand: u64,
+    weight: u32,
+}
+
+/// Weighted max-min fairness over `entries` with (at most) `capacity` to
+/// distribute. Deterministic: participants are processed in participant-id
+/// order and the result is a `BTreeMap`.
+///
+/// Loop: each round hands out proportional shares
+/// `share_i = capacity * weight_i / total_weight` (computed in `u128`,
+/// floored). Participants whose demand fits their share are satisfied and
+/// leave the active set together with their allocation; the rest re-share
+/// the remaining capacity. When nobody's demand fits, everyone gets its
+/// floor share and the integer-rounding remainder is distributed
+/// deterministically in participant-id order, one unit at a time, only to
+/// participants still below their demand.
+///
+/// Guarantees (asserted by tests): the sum of allocations never exceeds
+/// `capacity`; when every remaining demand is at least its computed
+/// share, the capacity is distributed in full.
+fn weighted_max_min(entries: Vec<Entry>, capacity: u64) -> BTreeMap<ParticipantId, u64> {
+    let mut result = BTreeMap::new();
+    let mut active: Vec<Entry> = Vec::new();
+
+    // Zero demand is trivially satisfied with an allocation of zero.
+    for entry in entries {
+        if entry.demand == 0 {
+            result.insert(entry.id, 0);
+        } else {
+            active.push(entry);
+        }
+    }
+    active.sort_by_key(|entry| entry.id);
+
+    let mut capacity = capacity;
+    while capacity > 0 && !active.is_empty() {
+        let total_weight: u128 = active.iter().map(|entry| u128::from(entry.weight)).sum();
+        let shares: Vec<u128> = active
+            .iter()
+            .map(|entry| u128::from(capacity) * u128::from(entry.weight) / total_weight)
+            .collect();
+
+        let mut satisfied = Vec::new();
+        for (index, entry) in active.iter().enumerate() {
+            if u128::from(entry.demand) <= shares[index] {
+                satisfied.push(index);
+            }
+        }
+
+        if satisfied.is_empty() {
+            // Every demand is above its share: hand out the floor shares,
+            // then spread the rounding remainder one unit at a time. The
+            // remainder is smaller than the number of active participants
+            // and every one of them still has room, so it is distributed
+            // completely.
+            for (index, entry) in active.iter().enumerate() {
+                result.insert(entry.id, to_u64(shares[index]));
+            }
+            let distributed: u128 = shares.iter().sum();
+            let mut remainder = u128::from(capacity) - distributed;
+            while remainder > 0 {
+                let mut handed_out = 0;
+                for entry in &active {
+                    if remainder == 0 {
+                        break;
+                    }
+                    if let Some(allocation) = result.get_mut(&entry.id) {
+                        if *allocation < entry.demand {
+                            *allocation += 1;
+                            remainder -= 1;
+                            handed_out += 1;
+                        }
+                    }
+                }
+                if handed_out == 0 {
+                    break;
+                }
+            }
+            active.clear();
+            break;
+        }
+
+        let allocated: u64 = satisfied.iter().map(|&index| active[index].demand).sum();
+        let satisfied_set: BTreeSet<usize> = satisfied.into_iter().collect();
+        let mut remaining = Vec::new();
+        for (index, entry) in active.into_iter().enumerate() {
+            if satisfied_set.contains(&index) {
+                result.insert(entry.id, entry.demand);
+            } else {
+                remaining.push(entry);
+            }
+        }
+        active = remaining;
+        capacity = capacity.saturating_sub(allocated);
+    }
+
+    // Capacity exhausted (or nothing to distribute): whoever is still
+    // active receives nothing.
+    for entry in active {
+        result.insert(entry.id, 0);
+    }
+    result
+}
+
+/// All values in the allocation arithmetic are bounded by the `u64`
+/// capacity, so this conversion cannot truncate; `u64::MAX` is a safe
+/// fallback that keeps the code panic-free.
+fn to_u64(value: u128) -> u64 {
+    u64::try_from(value).unwrap_or(u64::MAX)
+}
+
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::unwrap_used, clippy::expect_used)]
+
+    use std::num::{NonZeroU16, NonZeroU32};
+
+    use super::*;
+
+    fn id(value: u16) -> ParticipantId {
+        ParticipantId::new(NonZeroU16::new(value).unwrap())
+    }
+
+    fn contributing(participant: u16, weight: u32) -> Participant {
+        Participant::new(
+            id(participant),
+            Participation::Contributing {
+                weight: Weight::new(NonZeroU32::new(weight).unwrap()),
+            },
+        )
+    }
+
+    fn leeching(participant: u16) -> Participant {
+        Participant::new(id(participant), Participation::Leeching)
+    }
+
+    fn demand(participant: u16, rate: u64) -> Demand {
+        Demand::new(id(participant), Rate::new(rate))
+    }
+
+    fn capacity(value: u64) -> Capacity {
+        Capacity::try_new(value).unwrap()
+    }
+
+    fn rate_of(share: &FairShare, participant: u16) -> u64 {
+        share.rate_of(id(participant)).value()
+    }
+
+    #[test]
+    fn equal_demands_equal_weights_split_equally() {
+        let share = allocate(
+            capacity(800),
+            &[contributing(1, 1), contributing(2, 1)],
+            &[demand(1, 800), demand(2, 800)],
+        )
+        .unwrap();
+        assert_eq!(rate_of(&share, 1), 400);
+        assert_eq!(rate_of(&share, 2), 400);
+        assert_eq!(share.total_allocated().value(), 800);
+        assert_eq!(share.unused().value(), 0);
+    }
+
+    #[test]
+    fn small_demand_is_satisfied_first_then_remainder_is_redistributed() {
+        let share = allocate(
+            capacity(900),
+            &[contributing(1, 1), contributing(2, 1)],
+            &[demand(1, 100), demand(2, 900)],
+        )
+        .unwrap();
+        assert_eq!(rate_of(&share, 1), 100);
+        assert_eq!(rate_of(&share, 2), 800);
+    }
+
+    #[test]
+    fn weights_shape_the_split_two_to_one() {
+        let share = allocate(
+            capacity(900),
+            &[contributing(1, 2), contributing(2, 1)],
+            &[demand(1, 900), demand(2, 900)],
+        )
+        .unwrap();
+        assert_eq!(rate_of(&share, 1), 600);
+        assert_eq!(rate_of(&share, 2), 300);
+    }
+
+    #[test]
+    fn rounding_remainder_is_distributed_in_id_order() {
+        let share = allocate(
+            capacity(1001),
+            &[contributing(1, 1), contributing(2, 1)],
+            &[demand(1, 1000), demand(2, 1000)],
+        )
+        .unwrap();
+        // Shares are 500 + 500, the extra unit goes to the lower id.
+        assert_eq!(rate_of(&share, 1), 501);
+        assert_eq!(rate_of(&share, 2), 500);
+        assert_eq!(share.total_allocated().value(), 1001);
+    }
+
+    #[test]
+    fn saturated_contributor_leaves_nothing_for_the_leech() {
+        let share = allocate(
+            capacity(100),
+            &[contributing(1, 1), leeching(2)],
+            &[demand(1, 200), demand(2, 50)],
+        )
+        .unwrap();
+        assert_eq!(rate_of(&share, 1), 100);
+        assert_eq!(rate_of(&share, 2), 0);
+    }
+
+    #[test]
+    fn undersubscribed_contributor_leaves_leftover_for_the_leech() {
+        let share = allocate(
+            capacity(100),
+            &[contributing(1, 1), leeching(2)],
+            &[demand(1, 60), demand(2, 100)],
+        )
+        .unwrap();
+        assert_eq!(rate_of(&share, 1), 60);
+        assert_eq!(rate_of(&share, 2), 40);
+    }
+
+    #[test]
+    fn two_leeches_split_leftover_equally_capped_by_demand() {
+        let share = allocate(
+            capacity(100),
+            &[contributing(1, 1), leeching(2), leeching(3)],
+            &[demand(1, 50), demand(2, 100), demand(3, 20)],
+        )
+        .unwrap();
+        assert_eq!(rate_of(&share, 1), 50);
+        // Leftover 50: leech 3 (demand 20) is satisfied first, leech 2
+        // gets the rest.
+        assert_eq!(rate_of(&share, 2), 30);
+        assert_eq!(rate_of(&share, 3), 20);
+    }
+
+    #[test]
+    fn capacity_below_all_demands_yields_proportional_weighted_split() {
+        let share = allocate(
+            capacity(30),
+            &[contributing(1, 2), contributing(2, 1)],
+            &[demand(1, 100), demand(2, 100)],
+        )
+        .unwrap();
+        assert_eq!(rate_of(&share, 1), 20);
+        assert_eq!(rate_of(&share, 2), 10);
+        assert_eq!(share.unused().value(), 0);
+    }
+
+    #[test]
+    fn zero_demand_receives_zero_without_consuming_capacity() {
+        let share = allocate(
+            capacity(800),
+            &[contributing(1, 1), contributing(2, 1)],
+            &[demand(1, 0), demand(2, 500)],
+        )
+        .unwrap();
+        assert_eq!(rate_of(&share, 1), 0);
+        assert_eq!(share.allocations().get(&id(1)), Some(&Rate::new(0)));
+        assert_eq!(rate_of(&share, 2), 500);
+        assert_eq!(share.unused().value(), 300);
+    }
+
+    #[test]
+    fn empty_demands_allocates_nothing_and_everything_is_unused() {
+        let share = allocate(capacity(500), &[contributing(1, 1), leeching(2)], &[]).unwrap();
+        assert_eq!(share.total_allocated().value(), 0);
+        assert_eq!(share.unused().value(), 500);
+        assert!(share.allocations().values().all(|rate| rate.value() == 0));
+    }
+
+    #[test]
+    fn unknown_participant_in_demand_is_rejected() {
+        let error = allocate(capacity(100), &[contributing(1, 1)], &[demand(9, 50)]).unwrap_err();
+        assert!(matches!(error, FairnessError::UnknownParticipant(_)));
+    }
+
+    #[test]
+    fn duplicate_demand_is_rejected() {
+        let error = allocate(
+            capacity(100),
+            &[contributing(1, 1)],
+            &[demand(1, 50), demand(1, 20)],
+        )
+        .unwrap_err();
+        assert!(matches!(error, FairnessError::DuplicateDemand(_)));
+    }
+
+    #[test]
+    fn duplicate_participant_with_same_class_is_rejected() {
+        let error = allocate(
+            capacity(100),
+            &[contributing(1, 1), contributing(1, 1)],
+            &[],
+        )
+        .unwrap_err();
+        assert!(matches!(error, FairnessError::DuplicateParticipant(_)));
+    }
+
+    #[test]
+    fn duplicate_participant_with_different_class_is_rejected() {
+        let error = allocate(capacity(100), &[contributing(1, 1), leeching(1)], &[]).unwrap_err();
+        assert!(matches!(error, FairnessError::ConflictingParticipation(_)));
+    }
+
+    #[test]
+    fn all_demands_at_least_share_distributes_capacity_in_full() {
+        let share = allocate(
+            capacity(1000),
+            &[contributing(1, 1), contributing(2, 1), contributing(3, 2)],
+            &[demand(1, 1000), demand(2, 1000), demand(3, 1000)],
+        )
+        .unwrap();
+        assert_eq!(rate_of(&share, 1), 250);
+        assert_eq!(rate_of(&share, 2), 250);
+        assert_eq!(rate_of(&share, 3), 500);
+        assert_eq!(share.total_allocated().value(), 1000);
+        assert_eq!(share.unused().value(), 0);
+    }
+
+    /// Roadmap pilot scenario: two household downloads (contributing,
+    /// equal) plus one leech competing for the pooled uplinks.
+    #[test]
+    fn pilot_scenario_two_households_and_a_leech() {
+        let share = allocate(
+            capacity(600),
+            &[contributing(1, 1), contributing(2, 1), leeching(3)],
+            &[demand(1, 250), demand(2, 250), demand(3, 150)],
+        )
+        .unwrap();
+        assert_eq!(rate_of(&share, 1), 250);
+        assert_eq!(rate_of(&share, 2), 250);
+        assert_eq!(rate_of(&share, 3), 100);
+        assert_eq!(share.total_allocated().value(), 600);
+        assert_eq!(share.unused().value(), 0);
+    }
+}
@@ -0,0 +1,38 @@
+//! sn-fair — weighted max-min fairness scheduler for `SpiderNet`.
+//!
+//! Allocates a shared capacity (e.g. the pooled uplink bandwidth of a
+//! neighborhood) among participants with a two-class weighted max-min
+//! fairness scheme:
+//!
+//! 1. **Contributors** (households that offer capacity to the network,
+//!    with a configurable weight) are served first, via weighted max-min
+//!    fairness over the full capacity.
+//! 2. **Leeches** (receive-only households; explicitly allowed — the
+//!    network exists under neighbors, not against them) share whatever
+//!    the contributors leave unused, weighted max-min with implicit equal
+//!    weights, capped by their demand.
+//!
+//! Rationale: leeches have the lowest but still fair priority. When
+//! contributors saturate the capacity, leeches get (near) nothing;
+//! otherwise they benefit from the leftovers. No one is dropped from the
+//! system.
+//!
+//! Pure computation: no I/O, no async runtime.
+//!
+//! `SpiderNet` software carries no warranty and accepts no responsibility
+//! for its usage or for transported content (AGPL-3.0, sections 15/16 —
+//! responsibility for usage lies with the users).
+
+#![forbid(unsafe_code)]
+#![warn(clippy::pedantic)]
+// deny is the goal before the first release
+// PoC scope: doc-annotation noise, not correctness. Revisit at release.
+#![allow(clippy::must_use_candidate, clippy::missing_errors_doc)]
+
+pub mod error;
+pub mod fair;
+pub mod types;
+
+pub use error::FairnessError;
+pub use fair::{allocate, Allocation, Demand, FairShare, Participant, Participation};
+pub use types::{Capacity, ParticipantId, Rate, Weight};
@@ -0,0 +1,212 @@
+//! CLI for the sn-fair fairness scheduler.
+//!
+//! 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.
+//!
+//! Pilot scope: simulates how pooled neighborhood capacity is split —
+//! e.g. two household downloads plus a leech competing for the pooled
+//! uplinks.
+
+#![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)]
+
+use std::collections::BTreeMap;
+use std::num::NonZeroU32;
+use std::process::ExitCode;
+
+use clap::Parser;
+use sn_fair::fair::{Demand, Participant, Participation};
+use sn_fair::types::{Capacity, ParticipantId, Rate, Weight};
+use sn_fair::{allocate, FairShare};
+
+/// Parses `id=contributing[:weight]` or `id=leeching`.
+fn parse_participant_spec(spec: &str) -> Result<(ParticipantId, Participation), String> {
+    let (id_raw, class_raw) = spec.split_once('=').ok_or_else(|| {
+        format!("invalid participant spec `{spec}`, expected `id=contributing[:weight]` or `id=leeching`")
+    })?;
+    let participant = id_raw
+        .trim()
+        .parse::<ParticipantId>()
+        .map_err(|err| format!("invalid participant spec `{spec}`: {err}"))?;
+    let class = class_raw.trim();
+    let participation = match class.strip_prefix("contributing") {
+        Some(rest) => {
+            let weight = if rest.is_empty() {
+                Weight::default()
+            } else if let Some(weight_raw) = rest.strip_prefix(':') {
+                weight_raw
+                    .trim()
+                    .parse::<u32>()
+                    .ok()
+                    .and_then(NonZeroU32::new)
+                    .map(Weight::new)
+                    .ok_or_else(|| {
+                        format!("invalid weight `{weight_raw}` in `{spec}` (must be a non-zero number)")
+                    })?
+            } else {
+                return Err(format!(
+                    "invalid participation `{class}` in `{spec}`, expected `contributing[:weight]` or `leeching`"
+                ));
+            };
+            Participation::Contributing { weight }
+        }
+        None if class == "leeching" => Participation::Leeching,
+        None => {
+            return Err(format!(
+                "invalid participation `{class}` in `{spec}`, expected `contributing[:weight]` or `leeching`"
+            ))
+        }
+    };
+    Ok((participant, participation))
+}
+
+/// Parses `id=rate`.
+fn parse_demand_spec(spec: &str) -> Result<(ParticipantId, Rate), String> {
+    let (id_raw, rate_raw) = spec
+        .split_once('=')
+        .ok_or_else(|| format!("invalid demand spec `{spec}`, expected `id=rate`"))?;
+    let participant = id_raw
+        .trim()
+        .parse::<ParticipantId>()
+        .map_err(|err| format!("invalid demand spec `{spec}`: {err}"))?;
+    let rate = rate_raw.trim().parse::<u64>().map(Rate::new).map_err(|_| {
+        format!("invalid rate `{rate_raw}` in `{spec}` (must be a non-negative number)")
+    })?;
+    Ok((participant, rate))
+}
+
+#[derive(Debug, Parser)]
+#[command(
+    name = "sn-fair",
+    version,
+    about = "SpiderNet weighted max-min fairness scheduler (pilot)"
+)]
+struct Args {
+    /// Total pooled capacity, in Mbit/s.
+    #[arg(long)]
+    capacity: u64,
+
+    /// Participant: `id=contributing[:weight]` (offers its uplink,
+    /// weight shapes the share, default weight 1) or `id=leeching`
+    /// (receive-only). Repeatable.
+    #[arg(
+        long = "participant",
+        value_name = "ID=CLASS[:WEIGHT]",
+        value_parser = parse_participant_spec
+    )]
+    participants: Vec<(ParticipantId, Participation)>,
+
+    /// Demand: `id=rate` in Mbit/s. Repeatable; participants without a
+    /// demand receive nothing.
+    #[arg(
+        long = "demand",
+        value_name = "ID=RATE",
+        value_parser = parse_demand_spec
+    )]
+    demands: Vec<(ParticipantId, Rate)>,
+}
+
+fn run() -> Result<(), String> {
+    let args = Args::parse();
+    let capacity = Capacity::try_new(args.capacity).map_err(|err| err.to_string())?;
+
+    let participants: Vec<Participant> = args
+        .participants
+        .into_iter()
+        .map(|(id, participation)| Participant::new(id, participation))
+        .collect();
+    let demand_of: BTreeMap<ParticipantId, u64> = args
+        .demands
+        .iter()
+        .map(|(participant, rate)| (*participant, rate.value()))
+        .collect();
+    let demands: Vec<Demand> = args
+        .demands
+        .into_iter()
+        .map(|(participant, rate)| Demand::new(participant, rate))
+        .collect();
+
+    let share = allocate(capacity, &participants, &demands).map_err(|err| err.to_string())?;
+    print_report(capacity, &participants, &demand_of, &share);
+    Ok(())
+}
+
+fn print_report(
+    capacity: Capacity,
+    participants: &[Participant],
+    demand_of: &BTreeMap<ParticipantId, u64>,
+    share: &FairShare,
+) {
+    println!("NO WARRANTY — SpiderNet sn-fair (AGPL-3.0, no liability)");
+    println!();
+    println!("pooled capacity: {} Mbit/s", capacity.value());
+    println!();
+
+    let headers = ["participant", "class", "weight", "demand", "allocation"];
+    let mut rows = Vec::new();
+    for participant in participants {
+        let (class, weight) = match participant.participation() {
+            Participation::Contributing { weight } => ("contributing", weight.value().to_string()),
+            Participation::Leeching => ("leeching", "-".to_string()),
+        };
+        let demand = demand_of
+            .get(&participant.id())
+            .map_or_else(|| "0".to_string(), ToString::to_string);
+        rows.push([
+            participant.id().to_string(),
+            class.to_string(),
+            weight,
+            demand,
+            share.rate_of(participant.id()).to_string(),
+        ]);
+    }
+
+    let mut widths = [0usize; 5];
+    for (index, header) in headers.iter().enumerate() {
+        widths[index] = header.len();
+    }
+    for row in &rows {
+        for (index, cell) in row.iter().enumerate() {
+            widths[index] = widths[index].max(cell.len());
+        }
+    }
+
+    print_row(&headers.map(ToString::to_string), &widths);
+    for row in &rows {
+        print_row(row, &widths);
+    }
+
+    println!();
+    println!("total allocated: {}", share.total_allocated());
+    println!("unused:          {}", share.unused());
+}
+
+fn print_row(cells: &[String], widths: &[usize]) {
+    println!(
+        "{:<w0$}  {:<w1$}  {:>w2$}  {:>w3$}  {:>w4$}",
+        cells[0],
+        cells[1],
+        cells[2],
+        cells[3],
+        cells[4],
+        w0 = widths[0],
+        w1 = widths[1],
+        w2 = widths[2],
+        w3 = widths[3],
+        w4 = widths[4],
+    );
+}
+
+fn main() -> ExitCode {
+    match run() {
+        Ok(()) => ExitCode::SUCCESS,
+        Err(report) => {
+            eprintln!("NO WARRANTY — SpiderNet sn-fair (AGPL-3.0, no liability)");
+            eprintln!("error: {report}");
+            ExitCode::FAILURE
+        }
+    }
+}
@@ -0,0 +1,133 @@
+//! Domain types: newtypes over raw primitives, validated at construction.
+
+use std::fmt;
+use std::num::{NonZeroU16, NonZeroU32, NonZeroU64};
+use std::str::FromStr;
+
+use crate::error::FairnessError;
+
+/// Identifier of a participant (a household) in the fairness allocation.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct ParticipantId(NonZeroU16);
+
+impl ParticipantId {
+    pub const fn new(value: NonZeroU16) -> Self {
+        Self(value)
+    }
+
+    pub const fn value(self) -> u16 {
+        self.0.get()
+    }
+}
+
+impl fmt::Display for ParticipantId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0.get())
+    }
+}
+
+/// Parses a decimal id; rejects zero and non-numeric input.
+impl FromStr for ParticipantId {
+    type Err = FairnessError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let parsed = s.trim().parse::<u16>().ok().and_then(NonZeroU16::new);
+        parsed
+            .map(Self::new)
+            .ok_or_else(|| FairnessError::InvalidParticipantId(s.to_string()))
+    }
+}
+
+/// Relative share shaping of a contributing participant. 1 means equal
+/// share; a participant with weight 2 gets twice the share of one with
+/// weight 1 (as far as demands allow).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Weight(NonZeroU32);
+
+impl Weight {
+    pub const fn new(value: NonZeroU32) -> Self {
+        Self(value)
+    }
+
+    pub const fn value(self) -> u32 {
+        self.0.get()
+    }
+}
+
+impl Default for Weight {
+    fn default() -> Self {
+        Self(NonZeroU32::MIN)
+    }
+}
+
+/// A traffic rate, unit-agnostic: bytes per second, Mbit/s — whatever
+/// unit the caller works with (the CLI treats it as Mbit/s).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub struct Rate(u64);
+
+impl Rate {
+    pub const fn new(value: u64) -> Self {
+        Self(value)
+    }
+
+    pub const fn value(self) -> u64 {
+        self.0
+    }
+}
+
+impl fmt::Display for Rate {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+/// Total poolable capacity shared by all participants; strictly positive.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub struct Capacity(NonZeroU64);
+
+impl Capacity {
+    pub fn try_new(value: u64) -> Result<Self, FairnessError> {
+        NonZeroU64::new(value)
+            .map(Self)
+            .ok_or(FairnessError::ZeroCapacity)
+    }
+
+    pub const fn value(self) -> u64 {
+        self.0.get()
+    }
+}
+
+impl fmt::Display for Capacity {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0.get())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #![allow(clippy::unwrap_used, clippy::expect_used)]
+
+    use super::*;
+
+    #[test]
+    fn participant_id_parses_decimal_and_rejects_zero_and_garbage() {
+        assert_eq!("7".parse::<ParticipantId>().unwrap().value(), 7);
+        assert_eq!("  42 ".parse::<ParticipantId>().unwrap().value(), 42);
+        assert!("0".parse::<ParticipantId>().is_err());
+        assert!("abc".parse::<ParticipantId>().is_err());
+        assert!("-1".parse::<ParticipantId>().is_err());
+        assert!("70000".parse::<ParticipantId>().is_err());
+        assert!("".parse::<ParticipantId>().is_err());
+    }
+
+    #[test]
+    fn weight_defaults_to_one() {
+        assert_eq!(Weight::default().value(), 1);
+    }
+
+    #[test]
+    fn capacity_rejects_zero() {
+        assert!(Capacity::try_new(0).is_err());
+        assert_eq!(Capacity::try_new(5).unwrap().value(), 5);
+    }
+}