rustification-server/docs/realtime/issues/08-fan-out-architektur.md

# Fan-out-Architektur

Type: grilling
Status: resolved
Blocked by: —
Parent: ../map.md

## Question

Wie kommt ein Änderungs-Event von „Nutzer A hat geschrieben" zu allen betroffenen offenen
Verbindungen — und wie sieht die Naht aus, hinter der später `LISTEN/NOTIFY` steckt?

Entschieden ist bereits: Start mit In-Process-Fan-out, aber hinter einem Trait; `NOTIFY`
als zweite Implementierung; **kein Redis** (harte Regel: eine Binary + Postgres).

### Zu entscheiden

- **Registry-Struktur.** `conversation_id → Menge von Sendern`, `user_id → Menge von
  Verbindungen`, oder beides? Ein Nutzer kann in vielen Conversations sein und mehrere
  Geräte haben — welche Indizierung hält beides aus, ohne bei jedem Event über alles zu
  iterieren?
- **Kanaltyp.** `tokio::sync::broadcast` pro Conversation, `mpsc` pro Verbindung, oder
  ein globaler Broadcast mit clientseitigem Filter. Bei `broadcast` ist zu klären, wie
  `RecvError::Lagged` behandelt wird — es bedeutet hier „Resync", nicht „Fehler", und
  das muss explizit im Design stehen (Council-Hinweis des Engineer).
- **Auf- und Abräumen.** Wann wird ein Eintrag entfernt? Verbindungsabbrüche sind der
  Normalfall, nicht die Ausnahme — Leaks in der Registry sind der klassische Fehler.
- **Trait-Zuschnitt.** Wie schmal muss die Schnittstelle sein, damit die
  `NOTIFY`-Implementierung sie wirklich erfüllen kann? `NOTIFY` ist *fire and forget*
  ohne Zustellgarantie und ohne Rückkanal — ein Trait, das mehr verspricht als das,
  ist eine Lüge, die beim Wechsel auffliegt.
- **`NOTIFY`-Pfad konkret genug, um ihn zu bewerten.** Ein Channel pro Conversation
  (Postgres hat ein Limit an Channels pro Verbindung), oder ein Channel mit
  Conversation-ID im Payload? Wer hält die `LISTEN`-Verbindung, und was passiert, wenn
  sie abreißt?
- **Wo wird die `conversation_events`-Zeile geschrieben?** [*Monotoner Cursor*](02-monotoner-cursor.md)
  hat die Tabelle festgelegt, diese Frage aber bewusst hierher verschoben, weil sie mit
  `NOTIFY` zusammenfällt. Also: im Handler in derselben Transaktion, oder per
  Datenbank-Trigger? Ein Trigger feuert garantiert bei jedem Schreibweg, ein
  Handler-Aufruf ist einfacher zu verstehen. Bei Trigger + `NOTIFY` fällt die zweite
  Implementierung fast von selbst ab — das ist ein Argument, das geprüft gehört.
### Faktenlage aus der Auth-Recherche (14.08.2026)

Aus [`../research/03-ws-auth-reconnect.md`](../research/03-ws-auth-reconnect.md) §6, belegt:

- **`tokio::sync::watch` passt auffällig gut zum Dumb Pipe.** Es „retains only the last sent
  value" — was exakt die Semantik von „der Cursor der Conversation X ist jetzt N, hol dir den
  Rest per REST" ist. **Kein Lag-Fehler möglich**, weil Zwischenwerte per Konstruktion
  entbehrlich sind. Das ist ein ernstzunehmender Kandidat gegen `broadcast` und sollte hier
  gegen es antreten, nicht stillschweigend übergangen werden.
- **`RecvError::Lagged(n)` präzise:** `n` = Anzahl übersprungener Nachrichten, der Receiver
  bleibt **brauchbar**, sein Cursor springt auf die älteste noch vorhandene Nachricht.
  ⚠️ **„Lagged = Resync" ist nirgends als offizielles Muster dokumentiert** — gängige
  Community-Praxis ohne Primärquelle. Belegbar sind nur die Bausteine: `resubscribe()`
  (verwirft alles Alte), `Receiver::len()` als Frühwarnung *vor* dem Datenverlust, und
  `mpsc::try_send` pro Verbindung, um die Drop-Entscheidung **pro Client** statt global zu
  treffen.
- **`broadcast` speichert einmal und klont pro Receiver** ⇒ der Event-Typ sollte billig
  klonbar sein (`Arc<…>`, `bytes::Bytes`). Kapazität wird auf die nächste Zweierpotenz
  aufgerundet.
- **`async_broadcast` scheidet vermutlich aus:** es bietet echte Backpressure, womit ein
  langsamer WS-Client den **gesamten Producer** bremst (Head-of-Line über alle Clients) — das
  widerspricht der Map-Entscheidung „Best-effort + `?since`-Backfill" frontal.
  **`flume` scheidet sicher aus:** MPMC heißt dort Work-Queue, jede Nachricht geht an genau
  einen Receiver — kein Broadcast.
- **`PgListener` (sqlx)** reconnected selbst und re-subscribed alle Channels, belegt **aber
  eine dedizierte Postgres-Connection außerhalb des `PgPool`** (heute `max_connections = 10`)
  — beim Sizing einplanen. `NOTIFY` ist transaktional, feuert also erst bei COMMIT, was gut
  zur DB-als-Queue-Semantik passt. **Neue Gefahr:** wird die Notify-Queue voll (Listener
  hängen), schlagen Transaktionen **beim Commit fehl**. Und `NOTIFY` liefert keine Historie —
  ein verpasstes Signal ist weg, also dieselbe Resync-Frage eine Ebene tiefer.

- **Reihenfolge.** Garantiert das Fan-out die Reihenfolge der Events pro Conversation?
  Falls nein: ist das egal, weil der Cursor monoton ist und der Client ohnehin nachlädt?
  (Vermutlich ja — aber das explizit festhalten.)

---

## Answer

Entschieden am 14.08.2026.

### Vorbemerkung, die das ganze Ticket entschärft

Weil der Event `{conversation, seq}` ist und der Backfill über den Cursor läuft, zählt **nur
der jeweils neueste `seq` je Conversation**. 5, dann 6, dann 7 zuzustellen ist gleichwertig
damit, nur 7 zuzustellen. Das Fan-out darf zusammenfassen und sogar verlieren, ohne dass
Daten verlorengehen — schlimmstenfalls ist ein Client kurz veraltet, und das Manifest beim
nächsten Reconnect fängt alles auf.

Damit löst sich die `RecvError::Lagged`-Sorge aus
[*Recherche: WS-Auth, Heartbeat und Reconnect*](03-research-ws-auth-reconnect.md) weitgehend
auf: Lag ist hier kein Datenverlust, sondern Latenz.

### 1. Registry: `conversation_id → Verbindungen`, ein Kanal je Verbindung

Eine `HashMap<ConversationId, Set<Verbindung>>` im Serverzustand. Beim Verbindungsaufbau
werden die Mitgliedschaften einmal geladen und die Verbindung unter jeder eingetragen, beim
Trennen entfernt. Der Publisher schlägt einmal nach und stellt gezielt zu.

Entscheidend: **jede Verbindung hat genau einen Empfänger**. Die Verbindungsschleife bleibt
ein `select!` mit zwei Zweigen, statt über vierzig Kanäle gleichzeitig warten zu müssen. Die
Mitgliedschaftsliste braucht die Verbindung ohnehin zur Autorisierung — die Registry ist also
kein zusätzlicher Zustand.

Verworfen: ein Kanal je Conversation (eine Verbindung müsste auf N Empfängern warten,
`SelectAll` plus dynamisches An- und Abmelden während des Wartens) und ein globaler Kanal mit
Filter (zehn Zeilen Code, aber jede Verbindung wird bei jedem Event serverweit geweckt —
bei tausend Verbindungen und hundert Events pro Sekunde hunderttausend Aufwachvorgänge, die
fast alle nichts tun).

### 2. Zwei Registries, weil Mitgliedschaftswechsel anders adressiert werden

Ein Beitritt kann nicht über die Conversation-Registry laufen: der Nutzer ist dort noch nicht
eingetragen, das ist ja gerade der Punkt. Es gibt daher eine zweite Karte
`user_id → Verbindungen`. Mitgliedschafts-Events werden **pro Nutzer** zugestellt; die
Instanz, die diese Verbindung hält, trägt sie daraufhin in die Conversation-Registry ein oder
aus.

Damit funktioniert es auch über Instanzgrenzen: die Mitgliedschaftsänderung reist als `NOTIFY`
mit der `user_id` im Payload, und jede Instanz aktualisiert nur ihre eigenen Verbindungen.

### 3. Puffer je Verbindung: Dirty-Set mit `Notify`

Je Verbindung eine `HashMap<ConversationId, seq>` hinter einem Mutex plus ein `Notify`. Der
Publisher trägt ein und weckt; die Schreibtask leert die Karte und schickt einen Event je
Conversation.

Das kodiert „nur der neueste zählt" direkt im Datentyp: perfektes Zusammenfassen, kein
Verlust, Speicher begrenzt durch die Anzahl der Conversations statt durch die Event-Rate.
**Kein Überlauf, also kein Überlauf-Fehlerpfad.**

⚠️ **Wichtiges Detail:** der Eintrag ist `max(vorhandener, neuer)`, kein blindes `insert`.
`NOTIFY` garantiert Reihenfolge nur je Listener-Verbindung; ein blindes Überschreiben würde
sonst im ungünstigen Fall einen neueren `seq` durch einen älteren ersetzen.

Verworfen: `mpsc` + `try_send`. Bei vollem Puffer verwirft es den **neuen** Event und behält
die alten — also ausgerechnet den einzigen, auf den es ankommt, und im Log nicht von
Normalbetrieb unterscheidbar. Die Variante „bei Überlauf trennen" wäre korrekt, macht die
Puffergröße aber zur Stellschraube zwischen „trennt zu früh" und „merkt Blockade zu spät",
mit Pendelrisiko bei langsamen Mobilclients.

### 4. Log-Zeile im Handler, `NOTIFY` per Trigger auf `conversation_events`

Die Fachlogik bleibt in Rust: der Handler zieht den `seq`
(`UPDATE conversations … RETURNING`, was laut [*Monotoner Cursor*](02-monotoner-cursor.md)
ohnehin in seiner Transaktion passieren muss) und schreibt die Log-Zeile mit passendem `kind`.

Auf `conversation_events` sitzt ein Trigger, der **nichts** tut außer `pg_notify` zu feuern —
drei Zeilen PL/pgSQL ohne Fachlogik. Damit ist der Benachrichtigungspfad unvergesslich: wer
immer eine Log-Zeile schreibt, löst das Signal aus, auch künftige Handler und manuelle
Eingriffe. Und weil `NOTIFY` transaktional ist, feuert es erst beim Commit — genau dann, wenn
die Zeile sichtbar wird.

Verworfen: alles im Handler (jeder künftige Schreibweg müsste daran denken — die Sorte
Vergessen, die sich als „manchmal aktualisiert sich der Client nicht" äußert) und alles per
Trigger auf den Fachtabellen (die Kernlogik des Realtime-Layers läge dann in PL/pgSQL,
verteilt über drei Trigger, schlecht testbar und schwer zu lesen).

### 5. Kein Trait — immer über `NOTIFY`, auch bei einer Instanz

**Das revidiert eine Randbedingung der Map.** Sie sah „In-Process hinter einem Trait,
`NOTIFY` als zweite Implementierung" vor. Das wurde formuliert, bevor feststand, dass der
Event 60 Byte groß ist und der Trigger `NOTIFY` ohnehin feuert.

Der Server hört sich per `PgListener` selbst zu und speist die Registry. **Ein** Pfad für
Ein- und Mehr-Instanz-Betrieb: kein Trait, keine zweite Implementierung, und die
Mehr-Instanz-Fähigkeit ist ab Tag eins da statt nur vorgesehen. Der getestete Pfad ist
automatisch der Produktionspfad — der eigentliche Gewinn, denn beim Trait wäre ausgerechnet
der `NOTIFY`-Pfad beim ersten Skalierungsschritt der ungetestete gewesen.

Ein einziger Channel mit der Conversation-ID im Payload, nicht einer je Conversation:
dynamisches `LISTEN`/`UNLISTEN` über alle Conversations wäre unbegrenzt, und 60 Byte liegen
weit unter der 8-kB-Grenze.

**Betriebliche Kosten, bewusst akzeptiert:**

- Jeder Event läuft durch Postgres — auf demselben Host im Sub-Millisekunden-Bereich.
- `PgListener` belegt eine **dedizierte Verbindung außerhalb des Pools** (heute
  `max_connections = 10`) — beim Sizing einplanen.
- Hängt ein Listener und die Notify-Queue läuft voll, **scheitern Transaktionen beim
  Commit**. Das braucht Überwachung und gehört in die Fog-Notiz zur Observability.

⚠️ **Lücke, die diese Entscheidung einführt:** `PgListener` reconnected laut sqlx-Doku selbst,
aber Events während der Unterbrechung sind **weg** — `NOTIFY` hat keine Historie. Regel
daher: **nach jedem Listener-Reconnect schickt der Server allen Verbindungen einen
Resync-Hinweis**, woraufhin die Clients das Manifest neu ziehen. Das ist billig, selten, und
schließt die Lücke vollständig.

### 6. Auf- und Abräumen

- **Verbindungsaufbau:** Mitgliedschaften laden, in beide Registries eintragen.
- **Verbindungsabbau:** aus beiden entfernen. Abbrüche sind der Normalfall, nicht die
  Ausnahme — Leaks in der Registry sind hier der klassische Fehler.
- **Mitgliedschaftswechsel während der Verbindung:** über die `user_id`-Registry, siehe §2.
- Eine leere Conversation-Menge wird aus der Karte entfernt, damit sie nicht unbegrenzt wächst.

### 7. Reihenfolge

Das Fan-out garantiert **keine** Reihenfolge über Instanzen hinweg und braucht sie auch
nicht: der Cursor ist monoton, der Client lädt ohnehin den aktuellen Zustand, und das
Dirty-Set nimmt das Maximum. Ein zu früh oder zu spät zugestellter Event kann nichts
beschädigen — er kann nur überflüssig sein.