diff --git a/amethyst/src/main/res/values-cs-rCZ/strings.xml b/amethyst/src/main/res/values-cs-rCZ/strings.xml index 9caae76f1..4a49d92a2 100644 --- a/amethyst/src/main/res/values-cs-rCZ/strings.xml +++ b/amethyst/src/main/res/values-cs-rCZ/strings.xml @@ -396,6 +396,65 @@ Přesunout vše do nových záložek Záložky úspěšně přesunuty Koncepty + Naplánované příspěvky + Naplánovat + Naplánovaný čas + Příspěvky se publikují přibližně do 15 minut od naplánovaného času. + Vyberte naplánovaný čas + Naplánovat na… + Publikováno za %1$s + Mělo být před %1$s + Čas + Naplánovat příspěvek + Zrušit plánování + Trvalá oznámení vypnuta + Naplánované příspěvky se nemusí publikovat, dokud aplikaci znovu neotevřete. Pro spolehlivé plánování na pozadí povolte trvalá oznámení v Nastavení → Předvolby UI. + Naplánované příspěvky se nemusí publikovat, dokud aplikaci znovu neotevřete. Naplánované příspěvky jiných účtů se nespustí, dokud je aktivní tento účet. Pro spolehlivé plánování na pozadí povolte trvalá oznámení v Nastavení → Předvolby UI. + Za 1 hodinu + Zítra v 9:00 + Příští pondělí v 9:00 + Povolit trvalá oznámení? + Naplánované příspěvky se spolehlivě publikují pouze tehdy, když jsou povolena trvalá oznámení. Jinak se nemusí spustit, dokud aplikaci znovu neotevřete. + Otevřít nastavení + Přesto pokračovat + %1$s · za %2$s + %1$s · před %2$s + Zítra + Odhlášeno + Odhlášeno · smazáno %1$d naplánovaných příspěvků + Naplánovaný příspěvek publikován + Naplánovaný příspěvek selhal + Naplánované příspěvky + Oznámení, když je naplánovaný příspěvek publikován nebo selže při publikaci. + Odeslat hned + Žádné naplánované příspěvky + Napište poznámku a klepněte na ikonu hodin pro naplánování na později. + Chyba: %1$s + Naplánováno + Odesílá se… + Selhalo + Odesláno + Zrušeno + Máte %1$d naplánovaných příspěvků, které ještě nebyly publikovány. Odhlášením budou trvale smazány. + + %d ve frontě + %d ve frontě + %d ve frontě + %d ve frontě + + + · 1 do 1 h + · %d do 1 h + · %d do 1 h + · %d do 1 h + + + na 1 relay + na %d relaye + na %d relayů + na %d relayů + + ID příspěvku zkopírováno Ankety Otevřené Uzavřené @@ -2440,63 +2499,4 @@ Veřejná emoji jsou viditelná pro všechny a objeví se ve vaší nabídce reakcí a v automatickém doplňování \":\", když je tento balíček ve vašem seznamu emoji. Soukromá emoji jsou šifrovaně uložena na relayích a viditelná pouze pro vás. Objeví se ve vaší nabídce reakcí a v automatickém doplňování \":\" stejně jako veřejná. Gif - Naplánované příspěvky - Naplánovat - Naplánovaný čas - Příspěvky se publikují přibližně do 15 minut od naplánovaného času. - Vyberte naplánovaný čas - Naplánovat na… - Publikováno za %1$s - Mělo být před %1$s - Čas - Naplánovat příspěvek - Zrušit plánování - Trvalá oznámení vypnuta - Naplánované příspěvky se nemusí publikovat, dokud aplikaci znovu neotevřete. Pro spolehlivé plánování na pozadí povolte trvalá oznámení v Nastavení → Předvolby UI. - Naplánované příspěvky se nemusí publikovat, dokud aplikaci znovu neotevřete. Naplánované příspěvky jiných účtů se nespustí, dokud je aktivní tento účet. Pro spolehlivé plánování na pozadí povolte trvalá oznámení v Nastavení → Předvolby UI. - Za 1 hodinu - Zítra v 9:00 - Příští pondělí v 9:00 - Povolit trvalá oznámení? - Naplánované příspěvky se spolehlivě publikují pouze tehdy, když jsou povolena trvalá oznámení. Jinak se nemusí spustit, dokud aplikaci znovu neotevřete. - Otevřít nastavení - Přesto pokračovat - %1$s · za %2$s - %1$s · před %2$s - Zítra - Odhlášeno - Odhlášeno · smazáno %1$d naplánovaných příspěvků - Odeslat hned - Žádné naplánované příspěvky - Napište poznámku a klepněte na ikonu hodin pro naplánování na později. - Chyba: %1$s - Naplánováno - Odesílá se… - Selhalo - Odesláno - Zrušeno - Máte %1$d naplánovaných příspěvků, které ještě nebyly publikovány. Odhlášením budou trvale smazány. - Naplánovaný příspěvek publikován - Naplánovaný příspěvek selhal - Naplánované příspěvky - Oznámení, když je naplánovaný příspěvek publikován nebo selže při publikaci. - - %d ve frontě - %d ve frontě - %d ve frontě - %d ve frontě - - - · 1 do 1 h - · %d do 1 h - · %d do 1 h - · %d do 1 h - - - na 1 relay - na %d relaye - na %d relayů - na %d relayů - - ID příspěvku zkopírováno diff --git a/amethyst/src/main/res/values-de-rDE/strings.xml b/amethyst/src/main/res/values-de-rDE/strings.xml index 1f17c0c97..efed91e46 100644 --- a/amethyst/src/main/res/values-de-rDE/strings.xml +++ b/amethyst/src/main/res/values-de-rDE/strings.xml @@ -402,6 +402,58 @@ anz der Bedingungen ist erforderlich Alle in neue Lesezeichen verschieben Lesezeichen erfolgreich migriert Entwürfe + Geplante Beiträge + Planen + Geplante Zeit + Beiträge werden innerhalb von ~15 Minuten nach der geplanten Zeit veröffentlicht. + Geplante Zeit auswählen + Planen für… + Veröffentlicht in %1$s + Fällig vor %1$s + Zeit + Beitrag planen + Planung abbrechen + Dauerbenachrichtigungen deaktiviert + Geplante Beiträge werden möglicherweise erst veröffentlicht, wenn du die App das nächste Mal öffnest. Aktiviere Dauerbenachrichtigungen in Einstellungen → UI-Einstellungen für zuverlässige Hintergrundplanung. + Geplante Beiträge werden möglicherweise erst veröffentlicht, wenn du die App wieder öffnest. Geplante Beiträge anderer Konten werden nicht ausgelöst, solange dieses Konto aktiv ist. Aktiviere Dauerbenachrichtigungen in Einstellungen → UI-Einstellungen für zuverlässige Hintergrundplanung. + In 1 Stunde + Morgen 9 Uhr + Nächsten Montag 9 Uhr + Dauerbenachrichtigungen aktivieren? + Geplante Beiträge werden zuverlässig nur veröffentlicht, wenn Dauerbenachrichtigungen aktiviert sind. Andernfalls werden sie möglicherweise erst beim nächsten Öffnen der App ausgelöst. + Einstellungen öffnen + Trotzdem fortfahren + %1$s · vor %2$s + Morgen + Abgemeldet + Abgemeldet · %1$d geplante(n) Beitrag/Beiträge gelöscht + Geplanter Beitrag veröffentlicht + Geplanter Beitrag fehlgeschlagen + Geplante Beiträge + Benachrichtigungen, wenn ein geplanter Beitrag veröffentlicht wird oder die Veröffentlichung fehlschlägt. + Jetzt senden + Keine geplanten Beiträge + Verfasse eine Notiz und tippe auf das Uhr-Symbol, um sie für später zu planen. + Fehler: %1$s + Geplant + Wird gesendet… + Fehlgeschlagen + Gesendet + Abgebrochen + Du hast %1$d geplante(n) Beitrag/Beiträge, der/die noch nicht veröffentlicht wurden. Beim Abmelden werden sie dauerhaft gelöscht. + + %d in Warteschlange + %d in Warteschlange + + + · 1 fällig in 1 Std. + · %d fällig in 1 Std. + + + an 1 Relay + an %d Relays + + Beitrags-ID kopiert Umfragen Offen Geschlossen @@ -2431,57 +2483,4 @@ anz der Bedingungen ist erforderlich Öffentliche Emojis sind für alle sichtbar und erscheinen in deinem Reaktionsmenü und in der \":\"-Autovervollständigungsauswahl, wenn dieses Paket in deiner Emoji-Liste ist. Private Emojis werden verschlüsselt auf Relays gespeichert und sind nur für dich sichtbar. Sie erscheinen in deinem Reaktionsmenü und in der \":\"-Autovervollständigung wie öffentliche. Gif - Geplante Beiträge - Planen - Geplante Zeit - Beiträge werden innerhalb von ~15 Minuten nach der geplanten Zeit veröffentlicht. - Geplante Zeit auswählen - Planen für… - Veröffentlicht in %1$s - Fällig vor %1$s - Zeit - Beitrag planen - Planung abbrechen - Dauerbenachrichtigungen deaktiviert - Geplante Beiträge werden möglicherweise erst veröffentlicht, wenn du die App das nächste Mal öffnest. Aktiviere Dauerbenachrichtigungen in Einstellungen → UI-Einstellungen für zuverlässige Hintergrundplanung. - Geplante Beiträge werden möglicherweise erst veröffentlicht, wenn du die App wieder öffnest. Geplante Beiträge anderer Konten werden nicht ausgelöst, solange dieses Konto aktiv ist. Aktiviere Dauerbenachrichtigungen in Einstellungen → UI-Einstellungen für zuverlässige Hintergrundplanung. - In 1 Stunde - Morgen 9 Uhr - Nächsten Montag 9 Uhr - Dauerbenachrichtigungen aktivieren? - Geplante Beiträge werden zuverlässig nur veröffentlicht, wenn Dauerbenachrichtigungen aktiviert sind. Andernfalls werden sie möglicherweise erst beim nächsten Öffnen der App ausgelöst. - Einstellungen öffnen - Trotzdem fortfahren - %1$s · in %2$s - %1$s · vor %2$s - Morgen - Abgemeldet - Abgemeldet · %1$d geplante(n) Beitrag/Beiträge gelöscht - Jetzt senden - Keine geplanten Beiträge - Verfasse eine Notiz und tippe auf das Uhr-Symbol, um sie für später zu planen. - Fehler: %1$s - Geplant - Wird gesendet… - Fehlgeschlagen - Gesendet - Abgebrochen - Du hast %1$d geplante(n) Beitrag/Beiträge, der/die noch nicht veröffentlicht wurden. Beim Abmelden werden sie dauerhaft gelöscht. - Geplanter Beitrag veröffentlicht - Geplanter Beitrag fehlgeschlagen - Geplante Beiträge - Benachrichtigungen, wenn ein geplanter Beitrag veröffentlicht wird oder die Veröffentlichung fehlschlägt. - - %d in Warteschlange - %d in Warteschlange - - - · 1 fällig in 1 Std. - · %d fällig in 1 Std. - - - an 1 Relay - an %d Relays - - Beitrags-ID kopiert diff --git a/amethyst/src/main/res/values-pt-rBR/strings.xml b/amethyst/src/main/res/values-pt-rBR/strings.xml index 52702d633..0d7b99816 100644 --- a/amethyst/src/main/res/values-pt-rBR/strings.xml +++ b/amethyst/src/main/res/values-pt-rBR/strings.xml @@ -396,6 +396,59 @@ Mover Tudo para Novos Favoritos Favoritos migrados com sucesso Rascunhos + Posts agendados + Agendar + Hora agendada + Posts são publicados em até ~15 minutos após o horário agendado. + Escolher horário agendado + Agendar para… + Publica em %1$s + Devia ter sido publicado há %1$s + Hora + Agendar post + Cancelar agendamento + Notificações sempre ativas desativadas + Posts agendados podem não ser publicados até você reabrir o app. Ative notificações sempre ativas em Configurações → Preferências de UI para agendamento confiável em segundo plano. + Posts agendados podem não ser publicados até você reabrir o app. Posts agendados de outras contas não serão disparados enquanto esta conta estiver ativa. Ative notificações sempre ativas em Configurações → Preferências de UI para agendamento confiável em segundo plano. + Em 1 hora + Amanhã às 9h + Próxima segunda às 9h + Ativar notificações sempre ativas? + Posts agendados publicam de forma confiável apenas quando notificações sempre ativas estão ativadas. Caso contrário, podem não disparar até você reabrir o app. + Abrir configurações + Continuar mesmo assim + %1$s · em %2$s + %1$s · há %2$s + Amanhã + Desconectado + Desconectado · %1$d post(s) agendado(s) excluído(s) + Post agendado publicado + Post agendado falhou + Posts agendados + Notificações quando um post agendado é publicado ou falha ao publicar. + Enviar agora + Sem posts agendados + Componha uma nota e toque no ícone do relógio para agendá-la. + Erro: %1$s + Agendado + Enviando… + Falhou + Enviado + Cancelado + Você tem %1$d post(s) agendado(s) que ainda não foram publicados. Sair excluirá esses posts permanentemente. + + %d na fila + %d na fila + + + · 1 em 1h + · %d em 1h + + + para 1 relay + para %d relays + + ID do post copiado Enquetes Abertas Encerradas @@ -2426,57 +2479,4 @@ Emojis públicos são visíveis para todos e aparecem no seu menu de reações e no seletor de autocompletar \":\" quando este pacote está na sua lista de emojis. Emojis privados são armazenados criptografados em relays e visíveis apenas para você. Eles aparecem no seu menu de reações e no autocompletar \":\" assim como os públicos. Gif - Posts agendados - Agendar - Hora agendada - Posts são publicados em até ~15 minutos após o horário agendado. - Escolher horário agendado - Agendar para… - Publica em %1$s - Devia ter sido publicado há %1$s - Hora - Agendar post - Cancelar agendamento - Notificações sempre ativas desativadas - Posts agendados podem não ser publicados até você reabrir o app. Ative notificações sempre ativas em Configurações → Preferências de UI para agendamento confiável em segundo plano. - Posts agendados podem não ser publicados até você reabrir o app. Posts agendados de outras contas não serão disparados enquanto esta conta estiver ativa. Ative notificações sempre ativas em Configurações → Preferências de UI para agendamento confiável em segundo plano. - Em 1 hora - Amanhã às 9h - Próxima segunda às 9h - Ativar notificações sempre ativas? - Posts agendados publicam de forma confiável apenas quando notificações sempre ativas estão ativadas. Caso contrário, podem não disparar até você reabrir o app. - Abrir configurações - Continuar mesmo assim - %1$s · em %2$s - %1$s · há %2$s - Amanhã - Desconectado - Desconectado · %1$d post(s) agendado(s) excluído(s) - Enviar agora - Sem posts agendados - Componha uma nota e toque no ícone do relógio para agendá-la. - Erro: %1$s - Agendado - Enviando… - Falhou - Enviado - Cancelado - Você tem %1$d post(s) agendado(s) que ainda não foram publicados. Sair excluirá esses posts permanentemente. - Post agendado publicado - Post agendado falhou - Posts agendados - Notificações quando um post agendado é publicado ou falha ao publicar. - - %d na fila - %d na fila - - - · 1 em 1h - · %d em 1h - - - para 1 relay - para %d relays - - ID do post copiado diff --git a/amethyst/src/main/res/values-sv-rSE/strings.xml b/amethyst/src/main/res/values-sv-rSE/strings.xml index e396c3607..cfdc0ef45 100644 --- a/amethyst/src/main/res/values-sv-rSE/strings.xml +++ b/amethyst/src/main/res/values-sv-rSE/strings.xml @@ -396,6 +396,59 @@ Flytta allt till nya bokmärken Bokmärken migrerade Utkast + Schemalagda inlägg + Schemalägg + Schemalagd tid + Inlägg publiceras inom ~15 minuter från den schemalagda tiden. + Välj schemalagd tid + Schemalägg för… + Publiceras om %1$s + Skulle ha publicerats för %1$s sedan + Tid + Schemalägg inlägg + Avbryt schemaläggning + Alltid-på-aviseringar avstängda + Schemalagda inlägg kanske inte publiceras förrän du öppnar appen igen. Aktivera alltid-på i Inställningar → UI-inställningar för pålitlig schemaläggning i bakgrunden. + Schemalagda inlägg kanske inte publiceras förrän du öppnar appen igen. Andra kontons schemalagda inlägg utlöses inte medan detta konto är aktivt. Aktivera alltid-på i Inställningar → UI-inställningar för pålitlig schemaläggning i bakgrunden. + Om 1 timme + Imorgon kl. 09:00 + Nästa måndag kl. 09:00 + Aktivera alltid-på-aviseringar? + Schemalagda inlägg publiceras pålitligt endast när alltid-på-aviseringar är aktiverade. Annars kanske de inte utlöses förrän du öppnar appen igen. + Öppna inställningar + Fortsätt ändå + %1$s · om %2$s + %1$s · för %2$s sedan + Imorgon + Utloggad + Utloggad · %1$d schemalagda inlägg raderade + Schemalagt inlägg publicerat + Schemalagt inlägg misslyckades + Schemalagda inlägg + Aviseringar när ett schemalagt inlägg publiceras eller misslyckas att publicera. + Skicka nu + Inga schemalagda inlägg + Skriv en anteckning och tryck på klockikonen för att schemalägga den. + Fel: %1$s + Schemalagd + Skickar… + Misslyckades + Skickat + Avbrutet + Du har %1$d schemalagda inlägg som inte har publicerats än. Att logga ut raderar dem permanent. + + %d i kö + %d i kö + + + · 1 inom 1 h + · %d inom 1 h + + + till 1 relä + till %d reläer + + Inläggs-ID kopierat Omröstningar Öppna Stängda @@ -2425,57 +2478,4 @@ Offentliga emojis är synliga för alla och visas i din reaktionsmeny och i \":\"-autokompletteringen när detta paket finns i din emoji-lista. Privata emojis lagras krypterade på relän och är endast synliga för dig. De visas i din reaktionsmeny och i autokomplettering med \":\" precis som offentliga. Gif - Schemalagda inlägg - Schemalägg - Schemalagd tid - Inlägg publiceras inom ~15 minuter från den schemalagda tiden. - Välj schemalagd tid - Schemalägg för… - Publiceras om %1$s - Skulle ha publicerats för %1$s sedan - Tid - Schemalägg inlägg - Avbryt schemaläggning - Alltid-på-aviseringar avstängda - Schemalagda inlägg kanske inte publiceras förrän du öppnar appen igen. Aktivera alltid-på i Inställningar → UI-inställningar för pålitlig schemaläggning i bakgrunden. - Schemalagda inlägg kanske inte publiceras förrän du öppnar appen igen. Andra kontons schemalagda inlägg utlöses inte medan detta konto är aktivt. Aktivera alltid-på i Inställningar → UI-inställningar för pålitlig schemaläggning i bakgrunden. - Om 1 timme - Imorgon kl. 09:00 - Nästa måndag kl. 09:00 - Aktivera alltid-på-aviseringar? - Schemalagda inlägg publiceras pålitligt endast när alltid-på-aviseringar är aktiverade. Annars kanske de inte utlöses förrän du öppnar appen igen. - Öppna inställningar - Fortsätt ändå - %1$s · om %2$s - %1$s · för %2$s sedan - Imorgon - Utloggad - Utloggad · %1$d schemalagda inlägg raderade - Skicka nu - Inga schemalagda inlägg - Skriv en anteckning och tryck på klockikonen för att schemalägga den. - Fel: %1$s - Schemalagd - Skickar… - Misslyckades - Skickat - Avbrutet - Du har %1$d schemalagda inlägg som inte har publicerats än. Att logga ut raderar dem permanent. - Schemalagt inlägg publicerat - Schemalagt inlägg misslyckades - Schemalagda inlägg - Aviseringar när ett schemalagt inlägg publiceras eller misslyckas att publicera. - - %d i kö - %d i kö - - - · 1 inom 1 h - · %d inom 1 h - - - till 1 relä - till %d reläer - - Inläggs-ID kopierat diff --git a/geode/config.example.toml b/geode/config.example.toml index 5835a5527..3d706999b 100644 --- a/geode/config.example.toml +++ b/geode/config.example.toml @@ -25,6 +25,20 @@ contact = "admin@example.com" host = "0.0.0.0" port = 7447 path = "/" +# Ktor CIO event-loop pool sizing. Leave commented-out for sensible +# per-CPU defaults (typical for <2k concurrent connections). Lift on +# big-VM deployments targeting 10k+ connections — over-threading at +# low connection counts hurts L1/L2 cache locality, so always +# benchmark before/after when tuning these. +# +# Operators targeting >1k concurrent WebSockets should also raise the +# OS file-descriptor limit: `ulimit -n 65536` (or higher) before +# launching, plus a matching `LimitNOFILE=` in any systemd unit. The +# default of 1024 on most distros caps the relay well below 1k FDs +# (one per WS plus DB and listening sockets). +# connection_group_size = 4 +# worker_group_size = 16 +# call_group_size = 64 [database] # True keeps an in-memory SQLite db (events vanish on restart). Useful diff --git a/geode/plans/2026-05-07-connection-scaling.md b/geode/plans/2026-05-07-connection-scaling.md index fb1f8b353..e16ab3f11 100644 --- a/geode/plans/2026-05-07-connection-scaling.md +++ b/geode/plans/2026-05-07-connection-scaling.md @@ -1,5 +1,14 @@ # Connection scaling: pushing past 2 000 +> **Status (2026-05-07):** Sketches A and B shipped on +> `claude/connection-scaling-plan-YVjc8`. Sketch C landed as a smaller +> slice in Quartz — the streaming-filter cut — once the audit showed +> the rest of the plan's premise was overstated. Verification +> benchmarks (`connectionsHeldOpen10k`, `connectionsHeldOpenWithFanout`) +> are wired up but only run under `-DrunLoadBenchmark=true`. Remaining +> open work, including fan-out de-duplication, is now tracked in +> [`live-broadcast-fanout-index.md`](./2026-05-07-live-broadcast-fanout-index.md). + ## Problem Current measurement (`LoadBenchmark.connectionsHeldOpen`): **~2 000 @@ -25,69 +34,169 @@ the channel array, even though most connections never fan out. ## Sketch -### A — adaptive outQueue capacity +### A — adaptive outQueue capacity ✅ shipped -Start every connection with `INITIAL_OUTGOING_BUFFER = 64`. When the -producer side trySends and we observe queue depth crossing a high-water -mark (e.g. 75% full), grow the channel up to `MAX_OUTGOING_BUFFER = -8192`. This is not how `kotlinx.coroutines.channels.Channel` is -structured (capacity is fixed at construction), so the implementation -is "swap in a wider channel under a per-session lock when watermark -trips" — drains the old, then routes new sends through the new. +> Original plan: start at `INITIAL_OUTGOING_BUFFER = 64` and swap to a +> wider channel under a per-session lock when a high-water mark +> trips. **Not how it shipped.** -Expected: 90% of connections never fan out, so they stay at 64 slots -× ~512 B per ref ≈ 32 KB. At 5 000 conns that's ~160 MB → ~5 MB. -Hot-fanout connections still get the 2 MB cap. - -### B — per-relay event-loop pool sizing - -Ktor CIO defaults to one event-loop thread per available CPU. -Beyond a few thousand connections, this becomes the bottleneck — and -none of geode's per-connection work is CPU-bound (it's mostly waiting -on incoming frames). Tune CIO via: +What actually shipped is the simpler alternative the original Risks +section called out: `Channel.UNLIMITED` plus an `AtomicInteger` +backlog cap. kotlinx.coroutines' `BufferedChannel` allocates segments +lazily, so an unlimited channel pays only the small head-segment cost +on idle connections — there is no preallocated buffer to scale. ```kotlin -embeddedServer(CIO, ...) { - connectionGroupSize = max(2, Runtime.getRuntime().availableProcessors() / 2) - workerGroupSize = max(4, Runtime.getRuntime().availableProcessors()) - callGroupSize = max(8, Runtime.getRuntime().availableProcessors() * 4) +private val outQueue = Channel(capacity = Channel.UNLIMITED) +private val outstanding = AtomicInteger(0) + +// producer side +val depth = outstanding.incrementAndGet() +if (depth > MAX_OUTGOING_BUFFER) { // 8192 + outstanding.decrementAndGet() + droppedForBackpressure = true + outQueue.close() // NIP-01: drop the conn + return@connect +} +val res = outQueue.trySend(json) +if (!res.isSuccess) outstanding.decrementAndGet() // closed concurrently + +// writer side +for (json in outQueue) { + ws.outgoing.send(Frame.Text(json)) + outstanding.decrementAndGet() } ``` -Expose these through `RelayConfig.NetworkSection` so an operator on a -big VM can lift them. +Memory characteristic the plan asked for is intact: idle connections +no longer reserve an 8 192-slot fixed buffer; hot fan-out connections +still get bounded at the same 2 MiB cap before the slow-client cutoff +fires. NIP-01 ordering is preserved (no silent drop — connection is +killed at the cap). -### C — reduce per-message JSON allocations +Implementation: `geode/.../server/WebSocketSessionPump.kt`. The +channel-swap approach was rejected because +`Channel.UNLIMITED` already gives the lazy-allocation behavior the +swap was simulating, with none of the swap's race surface. -`OptimizedJsonMapper.fromJsonToCommand` allocates a `JsonNode` tree per -incoming frame. At 10k connections with 1 msg/s each that's 10k tree -allocations/sec. Investigate streaming Jackson + reusing `ObjectMapper` -per session, or using kotlinx-serialization's lower-overhead path. +### B — per-relay event-loop pool sizing ✅ shipped -This is more of a quartz-level change than geode-specific, but -geode's load benchmark is the right place to measure it. +Three optional knobs added to `[network]` in `RelayConfig`: -## How to verify +```toml +[network] +host = "0.0.0.0" +port = 7447 +path = "/" +# connection_group_size = 4 +# worker_group_size = 16 +# call_group_size = 64 +``` -Add to `geode.perf.LoadBenchmark`: +Default is **`null` (Ktor default)** — no behavior change unless an +operator explicitly tunes them. The values are wired through +`LocalRelayServer` into the new `embeddedServer(factory = CIO, +rootConfig = serverConfig {…}, configure = {…})` overload (the +short-form `embeddedServer(factory, host, port) {…}` overload doesn't +expose CIO config). The auto-connector that the short form created +now has to be added explicitly via `connector { host = …; port = …}`. -- `connectionsHeldOpen10k` — opens 10 000 idle WebSocket connections; - asserts no FD exhaustion + RSS stays under 1 GB. -- `connectionsHeldOpenWithFanout` — 5 000 idle subscribers, - 10 EPS published; measures p99 fanout latency at scale. +`config.example.toml` documents the knobs and includes the operator +note that targeting >1k connections needs `ulimit -n 65536` (or +matching `LimitNOFILE=` in a systemd unit). -The current `connectionsHeldOpen` benchmark stays as the baseline -floor (~2 000 conns). +### C — reduce per-message JSON allocations ✅ partially shipped (in Quartz) -## Risks +> Original plan claim: "`OptimizedJsonMapper.fromJsonToCommand` +> allocates a `JsonNode` tree per incoming frame." **Overstated.** -- **Adaptive channel swap is fiddly**: drains under the producer's nose - must preserve OK ordering. A simpler alternative: keep capacity fixed, - but lazily allocate a small `ArrayDeque` only when the first - message is sent. Channels in kotlinx.coroutines do allocate up-front. -- **Bumping CIO group sizes can hurt**: more threads can mean worse - L1/L2 locality. Always benchmark before/after, don't trust - intuitive sizing. -- **OS-level FD limit**: per-process FD limit on Linux defaults to - 1024 in many environments. Document the `ulimit -n` requirement - for operators targeting >1k connections. +Audit of `quartz/.../jackson` showed the Command/Message envelope is +already streaming: + +| Path | Already streaming? | Tree alloc? | +| ----------------------- | ------------------ | -------------------------------------------------- | +| `MessageDeserializer` | yes | only for `COUNT` result (rare) | +| `CommandDeserializer` | yes | only for **filter sub-objects** in REQ/COUNT/NEG-OPEN | +| `EventDeserializer` | yes | none — `currentName().hashCode()` dispatch | +| `ManualFilterDeserializer` | **no** | `jp.codec.readTree(jp)` per filter | + +So the only relay-inbound tree allocation worth chasing was filter +parsing — the bulk of the per-frame allocations on a REQ-heavy +relay. + +What shipped: a streaming `ManualFilterDeserializer.fromJson(jp: +JsonParser)` modeled exactly on `EventDeserializer`. Token-loop with +field-name dispatch (`ids` / `authors` / `kinds` / `since` / `until` / +`limit` / `search`, plus dynamic `#x` / `&x` tag keys), and +`readStringArray` / `readIntArray` helpers that drop invalid entries +silently to match the tree path's `mapNotNull { asTextOrNull() }` +tolerance. Wired into all four internal call sites: +`FilterDeserializer.deserialize` and the three `CommandDeserializer` +paths (REQ, COUNT, NEG-OPEN). + +The tree-based `fromJson(ObjectNode)` overload is retained for +external/cross-format adapters (Quartz is a published library). + +What was NOT done — and why: +- **Streaming Jackson for the Command envelope**: already streaming. + No allocation to remove. +- **kotlinx-serialization for the inbound path**: not pursued. The + cross-mapper round-trip tests in `KotlinSerializationMapperTest` + show the two formats are interchangeable, but the engine swap is a + much larger lift than the filter cut and there's no evidence the + KS path is faster on this code shape. +- **Per-session `ObjectMapper`**: Jackson's `ObjectMapper` is + thread-safe and stateless — sharing one is the recommended pattern. + Per-session would *increase* allocation, not decrease it. + +## How to verify ✅ shipped + +Two new benchmarks in `geode.perf.LoadBenchmark`, gated behind +`-DrunLoadBenchmark=true`: + +- **`connectionsHeldOpen10k`** — opens 10 000 idle WebSocket + connections, asserts every one settles to EOSE inside 120 s, and + measures retained JVM heap (after `System.gc()` + 200 ms settle) + with a 1 GiB ceiling assertion. Requires `ulimit -n 32768` on + Linux. +- **`connectionsHeldOpenWithFanout`** — 5 000 subscribers all + matching `kinds:[1]`, one publisher emitting `targetEps × duration` + events, prints p50 / p99 last-fanout latency. No assertion on + latency — just regression-detection via stdout logging. + +The original `connectionsHeldOpen` benchmark stays as the **baseline +floor (~2 000 conns)** for before/after comparisons. + +Note on heap-vs-RSS: the original plan said "RSS stays under 1 GB" +but the JVM can only measure heap from inside; `Runtime.totalMemory +- freeMemory` is what the benchmark asserts on. RSS will be higher +because of code, native buffers, off-heap (Ktor CIO), etc. + +## Risks (post-implementation) + +- ~~**Adaptive channel swap is fiddly**~~ — sidestepped by using + `Channel.UNLIMITED` instead of swapping bounded channels. +- **Bumping CIO group sizes can hurt** — kept the defaults `null`. + Operators must opt in, and the docstrings explicitly say to + benchmark before/after. +- **OS-level FD limit** — documented in `config.example.toml` next to + the CIO knobs. Test prereq is also documented in the benchmark + KDoc. + +## Open work + +- **Fan-out de-duplication** — when one EVENT matches N subscribers, + we currently re-serialize and copy the JSON N times into N + channels. Caching one pre-serialized payload per event and + broadcasting a shared reference is a much bigger win than anything + in this plan; tracked in + [`live-broadcast-fanout-index.md`](./2026-05-07-live-broadcast-fanout-index.md). +- **Filter-matching index** — same plan. At 10k conns × ~5 filters + that's 50k evaluations per published EVENT, almost all of which + could be culled by indexing subscriptions on `kinds` / `authors` / + `#e` / `#p`. +- **Netty engine evaluation** — Ktor's Netty engine handles many idle + connections with measurably lower per-connection overhead than + CIO. Not pursued here because it changes the transport layer + wholesale; revisit only if the CIO knobs in (B) prove insufficient + for an operator at 20k+ connections. diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt index 3840bf482..56ef2f6da 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/LocalRelayServer.kt @@ -32,8 +32,10 @@ import io.ktor.http.ContentType import io.ktor.http.HttpHeaders import io.ktor.http.HttpStatusCode import io.ktor.server.application.install +import io.ktor.server.application.serverConfig import io.ktor.server.cio.CIO import io.ktor.server.cio.CIOApplicationEngine +import io.ktor.server.engine.connector import io.ktor.server.engine.embeddedServer import io.ktor.server.request.header import io.ktor.server.response.respondText @@ -109,6 +111,16 @@ class LocalRelayServer( * RPC payload. */ val maxAdminBodyBytes: Int = 1 shl 20, + /** + * Ktor CIO acceptor-thread count. `null` keeps Ktor's default. + * Lift on machines with many cores when targeting 10k+ + * concurrent connections — see `[network]` config docs. + */ + val connectionGroupSize: Int? = null, + /** Ktor CIO worker-thread count. `null` keeps Ktor's default. */ + val workerGroupSize: Int? = null, + /** Ktor CIO call-handling thread count. `null` keeps Ktor's default. */ + val callGroupSize: Int? = null, ) { private val infoHolder = object : Nip86Server.InfoHolder { @@ -167,51 +179,78 @@ class LocalRelayServer( * [url] is safe to read on the very next line. */ fun start(): LocalRelayServer { + // Snapshot the constructor-supplied overrides into locals so + // the `configure` lambda below can assign to its receiver + // without the names colliding with outer properties. + val connGrp = connectionGroupSize + val workGrp = workerGroupSize + val callGrp = callGroupSize + val bindHost = host + val bindPort = port val server = - embeddedServer(CIO, host = host, port = port) { - install(WebSockets) { - maxFrameBytes?.let { maxFrameSize = it } - } - routing { - // NIP-11: GET on the relay URL with Accept: - // application/nostr+json returns the relay info doc. - // We mount this *before* the webSocket route so Ktor - // serves NIP-11 for plain HTTP GETs and only upgrades - // to a WebSocket when the request is a WS upgrade. - get(path) { - val accept = call.request.header(HttpHeaders.Accept).orEmpty() - if (accept.contains("application/nostr+json")) { - call.response.headers.append("Access-Control-Allow-Origin", "*") - call.respondText( - relay.info.json, - ContentType.parse("application/nostr+json"), - ) - } else { - call.respondText( - "Use a Nostr client (NIP-01 WebSocket) or send Accept: application/nostr+json (NIP-11).", - ContentType.Text.Plain, - HttpStatusCode.UpgradeRequired, - ) + embeddedServer( + factory = CIO, + rootConfig = + serverConfig { + module { + install(WebSockets) { + maxFrameBytes?.let { maxFrameSize = it } + } + routing { + // NIP-11: GET on the relay URL with Accept: + // application/nostr+json returns the relay info doc. + // We mount this *before* the webSocket route so Ktor + // serves NIP-11 for plain HTTP GETs and only upgrades + // to a WebSocket when the request is a WS upgrade. + get(path) { + val accept = call.request.header(HttpHeaders.Accept).orEmpty() + if (accept.contains("application/nostr+json")) { + call.response.headers.append("Access-Control-Allow-Origin", "*") + call.respondText( + relay.info.json, + ContentType.parse("application/nostr+json"), + ) + } else { + call.respondText( + "Use a Nostr client (NIP-01 WebSocket) or send Accept: application/nostr+json (NIP-11).", + ContentType.Text.Plain, + HttpStatusCode.UpgradeRequired, + ) + } + } + // NIP-86: POST application/nostr+json+rpc with a NIP-98 + // signed Authorization header → JSON-RPC dispatch. + post(path) { + nip86Route.handle(call) + } + webSocket(path) { + if (shuttingDown) { + // Just return — Ktor closes the WS for us. + return@webSocket + } + WebSocketSessionPump(this).pump( + server = relay.server, + registerSession = activeSessions::add, + unregisterSession = activeSessions::remove, + ) + } + } } + }, + configure = { + connector { + host = bindHost + port = bindPort } - // NIP-86: POST application/nostr+json+rpc with a NIP-98 - // signed Authorization header → JSON-RPC dispatch. - post(path) { - nip86Route.handle(call) - } - webSocket(path) { - if (shuttingDown) { - // Just return — Ktor closes the WS for us. - return@webSocket - } - WebSocketSessionPump(this).pump( - server = relay.server, - registerSession = activeSessions::add, - unregisterSession = activeSessions::remove, - ) - } - } - } + // Keep Ktor defaults unless the operator overrode + // them — Ktor's per-CPU sizing is sensible for + // most deployments, and over-threading hurts L1/L2 + // locality at low connection counts. + connGrp?.let { connectionGroupSize = it } + workGrp?.let { workerGroupSize = it } + callGrp?.let { callGroupSize = it } + }, + ) server.start(wait = false) engine = server.engine // Ktor 3.x made resolvedConnectors() suspend. We block here so diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt index 228c40709..e7b7a97b0 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/Main.kt @@ -117,6 +117,9 @@ fun main(args: Array) { maxFrameBytes = frameLimit, adminPubkeys = config.admin.pubkeys.toSet(), publicUrl = config.admin.public_url, + connectionGroupSize = config.network.connection_group_size, + workerGroupSize = config.network.worker_group_size, + callGroupSize = config.network.call_group_size, ).start() Runtime.getRuntime().addShutdownHook( diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt index b1b99634d..1417ec61f 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/config/RelayConfig.kt @@ -96,6 +96,29 @@ data class RelayConfig( val host: String = "0.0.0.0", val port: Int = 7447, val path: String = "/", + /** + * Ktor CIO acceptor-thread count. `null` (default) keeps Ktor's + * default sizing — fine up to a few thousand concurrent + * connections. On big-VM deployments targeting 10k+ + * connections, lift this to roughly half the available cores + * so the acceptor doesn't starve workers. + */ + val connection_group_size: Int? = null, + /** + * Ktor CIO worker-thread count (handles socket I/O). `null` + * keeps Ktor's default. Each connection's WebSocket read/write + * is dispatched onto this pool; for many idle long-lived + * connections the pool can stay small, but 10k+ connections + * benefit from sizing this to the full CPU count. + */ + val worker_group_size: Int? = null, + /** + * Ktor CIO call-handling thread count. `null` keeps Ktor's + * default. Sized higher than [worker_group_size] because each + * call (incl. WebSocket upgrade) may suspend on I/O — at + * 10k+ connections, ~4× cores is a reasonable starting point. + */ + val call_group_size: Int? = null, ) data class DatabaseSection( diff --git a/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt b/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt index 7cd700a33..f5319779d 100644 --- a/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt +++ b/geode/src/main/kotlin/com/vitorpamplona/geode/server/WebSocketSessionPump.kt @@ -29,6 +29,7 @@ import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.ClosedSendChannelException import kotlinx.coroutines.channels.consumeEach import kotlinx.coroutines.launch +import java.util.concurrent.atomic.AtomicInteger /** * Per-WebSocket pump that owns the bounded outbound queue and the @@ -44,15 +45,39 @@ import kotlinx.coroutines.launch * 3. `finally`-style teardown closes the queue, cancels the * writer, unregisters the session, and closes it. * - * Slow-client policy: when [outQueue] fills, [SESSION_OUTGOING_BUFFER] - * frames behind, the connection is dropped rather than silently - * losing EVENT/EOSE — silent drop would corrupt NIP-01. + * Slow-client policy: once the outbound backlog reaches + * [MAX_OUTGOING_BUFFER] frames, the connection is dropped rather + * than silently losing EVENT/EOSE — silent drop would corrupt + * NIP-01. + * + * Memory model: the outbound queue is `Channel.UNLIMITED`, which in + * kotlinx.coroutines allocates segments lazily — an idle connection + * pays only a small head-segment cost. The cap is enforced via + * [outstanding] rather than the channel's own capacity so we don't + * reserve a fixed-size buffer up-front for every connection. At + * 5 000+ idle connections this matters: an 8 192-slot fixed buffer + * per connection would otherwise dominate JVM heap usage even + * though the vast majority of connections never fan out. */ internal class WebSocketSessionPump( private val ws: DefaultWebSocketServerSession, ) { - private val outQueue = Channel(capacity = SESSION_OUTGOING_BUFFER) - private var droppedForBackpressure = false + /** + * Unbounded channel — bounded by [outstanding] above, not by the + * channel's own capacity. See class kdoc for memory rationale. + */ + private val outQueue = Channel(capacity = Channel.UNLIMITED) + + /** + * Number of frames queued but not yet written to the socket. + * Producer increments before [Channel.trySend]; writer decrements + * after the frame is handed to Ktor. When this would cross + * [MAX_OUTGOING_BUFFER] we treat the client as slow and close + * the queue. + */ + private val outstanding = AtomicInteger(0) + + @Volatile private var droppedForBackpressure = false suspend fun pump( server: NostrServer, @@ -64,6 +89,7 @@ internal class WebSocketSessionPump( try { for (json in outQueue) { ws.outgoing.send(Frame.Text(json)) + outstanding.decrementAndGet() } } catch (_: ClosedSendChannelException) { // socket closed — outer handler runs normal teardown. @@ -71,13 +97,22 @@ internal class WebSocketSessionPump( } val session = server.connect { json -> - val res = outQueue.trySend(json) - if (!res.isSuccess && !res.isClosed) { - // Buffer is full → slow client. Mark + close the - // queue; the writer drains, then the outer handler - // closes the WS session. + // The channel itself is UNLIMITED, so trySend can't + // report "full". Enforce the cap explicitly: increment + // first, refuse if we'd cross the bound, otherwise + // enqueue. + val depth = outstanding.incrementAndGet() + if (depth > MAX_OUTGOING_BUFFER) { + outstanding.decrementAndGet() droppedForBackpressure = true outQueue.close() + return@connect + } + val res = outQueue.trySend(json) + if (!res.isSuccess) { + // Channel was closed concurrently (e.g. teardown). + // Roll back the counter; nothing more to do. + outstanding.decrementAndGet() } } registerSession(session) @@ -98,17 +133,19 @@ internal class WebSocketSessionPump( companion object { /** - * Per-session outbound buffer size. When a slow client falls + * Per-session outbound backlog cap. When a slow client falls * this many frames behind, we close their connection rather * than silently dropping further frames (which would corrupt * NIP-01 by missing EVENT/EOSE messages). * * Sized to hold fan-out for a connection holding several * thousand subscriptions when one event matches all of them - * — the realistic upper bound for a relay client. At ~250B - * per frame this caps per-session memory at ~2 MiB before - * we drop the connection. + * — the realistic upper bound for a relay client. At ~250 B + * per frame this caps per-session worst-case memory at + * ~2 MiB before we drop the connection. Idle connections + * pay only the small head-segment cost of an unlimited + * channel (≈ a few hundred bytes), not the full cap. */ - const val SESSION_OUTGOING_BUFFER: Int = 8192 + const val MAX_OUTGOING_BUFFER: Int = 8192 } } diff --git a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt index a083dd812..04a5867fc 100644 --- a/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt +++ b/geode/src/test/kotlin/com/vitorpamplona/geode/perf/LoadBenchmark.kt @@ -139,6 +139,195 @@ class LoadBenchmark { } } + /** + * Holds 10 000 idle WebSocket connections open against a single + * relay. Verifies that the adaptive outQueue (sketch A in + * [connection-scaling plan][1]) lets us cross the ~2 000-connection + * floor measured by [connectionsHeldOpen] without FD exhaustion or + * runaway RSS. + * + * RUN PREREQ: requires a process FD limit ≥ ~12 000 (each WS uses + * one FD on each side plus margin). On Linux: `ulimit -n 32768` + * before launching the test JVM. + * + * [1]: geode/plans/2026-05-07-connection-scaling.md + */ + @Test + fun connectionsHeldOpen10k() = + benchmark("connections held open 10k") { + val target = 10_000 + runBenchmarkServer { server, http -> + val httpUrl = + okhttp3.Request + .Builder() + .url(server.url.replace("ws://", "http://")) + .build() + val sockets = java.util.concurrent.CopyOnWriteArrayList() + val opened = AtomicLong() + val gotEose = AtomicLong() + val opens = + measureTime { + repeat(target) { + val ws = + http.newWebSocket( + httpUrl, + object : okhttp3.WebSocketListener() { + override fun onOpen( + webSocket: okhttp3.WebSocket, + response: okhttp3.Response, + ) { + opened.incrementAndGet() + webSocket.send( + """["REQ","s",{"kinds":[1],"limit":1}]""", + ) + } + + override fun onMessage( + webSocket: okhttp3.WebSocket, + text: String, + ) { + if (text.startsWith("[\"EOSE\"")) { + gotEose.incrementAndGet() + } + } + }, + ) + sockets += ws + } + val deadline = System.currentTimeMillis() + 120_000 + while (gotEose.get() < target && System.currentTimeMillis() < deadline) { + Thread.sleep(50) + } + } + // JVM heap usage, not OS RSS — we can only measure + // what the JVM itself has allocated. Force a GC first + // so the reading reflects retained bytes, not in-flight + // allocation churn from the connect ramp-up. + val rt = Runtime.getRuntime() + System.gc() + Thread.sleep(200) + val heapMb = (rt.totalMemory() - rt.freeMemory()) / (1024 * 1024) + println( + "target=$target opened=${opened.get()} eosed=${gotEose.get()} " + + "active=${server.activeSessionCount} elapsedMs=${opens.inWholeMilliseconds} " + + "heapMb=$heapMb", + ) + sockets.forEach { runCatching { it.cancel() } } + check(gotEose.get() == target.toLong()) { + "expected $target EOSE but got ${gotEose.get()} — connection scaling regression" + } + check(heapMb < 1024) { + "JVM heap $heapMb MiB exceeded 1 GiB ceiling for $target idle connections" + } + } + } + + /** + * 5 000 idle subscribers, one publisher emitting 10 EPS for 10 s. + * Measures fan-out latency at scale — exercises the queue path + * for a connection that *does* fan out, not just an idle one. + * + * Each subscriber matches every published event (`kinds:[1]`), + * so a single EVENT generates 5 000 outbound frames per tick. + */ + @Test + fun connectionsHeldOpenWithFanout() = + benchmark("connections held open with fanout") { + val subs = 5_000 + val durationSeconds = 10 + val targetEps = 10 + runBenchmarkServer { server, http -> + val scope = CoroutineScope(Dispatchers.Default + SupervisorJob()) + val subClient = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope) + val pubClient = NostrClient(BasicOkHttpWebSocket.Builder { _ -> http }, scope) + try { + val relayUrl = server.url.normalizeRelayUrl() + val received = AtomicLong() + val eosed = AtomicLong() + // Last-receive timestamp per event id. The N-th + // subscriber to deliver wins; combined with the + // publish timestamp this gives us the full fan-out + // duration to the slowest subscriber. + val lastReceiveNs = + java.util.concurrent.ConcurrentHashMap() + + repeat(subs) { i -> + subClient.subscribe( + "fanout-$i", + mapOf(relayUrl to listOf(Filter(kinds = listOf(1)))), + object : SubscriptionListener { + override fun onEvent( + event: com.vitorpamplona.quartz.nip01Core.core.Event, + isLive: Boolean, + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + lastReceiveNs + .computeIfAbsent(event.id) { AtomicLong() } + .set(System.nanoTime()) + received.incrementAndGet() + } + + override fun onEose( + relay: NormalizedRelayUrl, + forFilters: List?, + ) { + eosed.incrementAndGet() + } + }, + ) + } + + runBlocking { + withTimeout(120_000) { + while (eosed.get() < subs) kotlinx.coroutines.delay(100) + } + } + println("$subs subs ready; publishing ${targetEps * durationSeconds} events at $targetEps EPS...") + + val signer = NostrSignerSync(KeyPair()) + val publishedAt = java.util.concurrent.ConcurrentHashMap() + val totalEvents = targetEps * durationSeconds + val tickIntervalMs = 1000L / targetEps + + runBlocking { + repeat(totalEvents) { i -> + val event = signer.sign(TextNoteEvent.build("fanout-$i")) + publishedAt[event.id] = System.nanoTime() + pubClient.publishAndConfirm(event, setOf(relayUrl)) + kotlinx.coroutines.delay(tickIntervalMs) + } + } + + // Wait for fan-out completion (or 30s, whichever first). + runBlocking { + withTimeout(30_000) { + while (received.get() < subs.toLong() * totalEvents) { + kotlinx.coroutines.delay(100) + } + } + } + + val perEventLastMs = + lastReceiveNs.entries + .mapNotNull { (id, last) -> + publishedAt[id]?.let { (last.get() - it) / 1_000_000.0 } + }.sorted() + val p50 = perEventLastMs.getOrNull(perEventLastMs.size / 2) ?: -1.0 + val p99 = perEventLastMs.getOrNull((perEventLastMs.size * 99) / 100) ?: -1.0 + println( + "subs=$subs events=$totalEvents received=${received.get()}/${subs.toLong() * totalEvents} " + + "p50LastFanoutMs=${"%.1f".format(p50)} " + + "p99LastFanoutMs=${"%.1f".format(p99)}", + ) + } finally { + subClient.disconnect() + pubClient.disconnect() + scope.cancel() + } + } + } + /** * One publisher sends 10k events serially. Measures the round-trip * `EVENT` → `OK true` time, which is dominated by SQLite write diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toRelay/CommandDeserializer.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toRelay/CommandDeserializer.kt index 3b5a8d622..258e52c8d 100644 --- a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toRelay/CommandDeserializer.kt +++ b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/commands/toRelay/CommandDeserializer.kt @@ -24,7 +24,6 @@ import com.fasterxml.jackson.core.JsonParser import com.fasterxml.jackson.core.JsonToken import com.fasterxml.jackson.databind.DeserializationContext import com.fasterxml.jackson.databind.deser.std.StdDeserializer -import com.fasterxml.jackson.databind.node.ObjectNode import com.vitorpamplona.quartz.nip01Core.jackson.EventDeserializer import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter import com.vitorpamplona.quartz.nip01Core.relay.filters.ManualFilterDeserializer @@ -52,9 +51,9 @@ class CommandDeserializer : StdDeserializer(Command::class.java) { val filters = mutableListOf() while (jp.nextToken() != JsonToken.END_ARRAY) { - val filterObj: ObjectNode = jp.codec.readTree(jp) - val filter = ManualFilterDeserializer.fromJson(filterObj) - filters.add(filter) + // currentToken is now START_OBJECT for each filter; + // the streaming parser consumes through END_OBJECT. + filters.add(ManualFilterDeserializer.fromJson(jp)) } ReqCmd( @@ -68,9 +67,7 @@ class CommandDeserializer : StdDeserializer(Command::class.java) { val filters = mutableListOf() while (jp.nextToken() != JsonToken.END_ARRAY) { - val filterObj: ObjectNode = jp.codec.readTree(jp) - val filter = ManualFilterDeserializer.fromJson(filterObj) - filters.add(filter) + filters.add(ManualFilterDeserializer.fromJson(jp)) } CountCmd( @@ -102,9 +99,8 @@ class CommandDeserializer : StdDeserializer(Command::class.java) { NegOpenCmd.LABEL -> { val subId = jp.nextTextValue() - jp.nextToken() - val filterObj: ObjectNode = jp.codec.readTree(jp) - val filter = ManualFilterDeserializer.fromJson(filterObj) + jp.nextToken() // advance to filter's START_OBJECT + val filter = ManualFilterDeserializer.fromJson(jp) val initialMessage = jp.nextTextValue() NegOpenCmd( diff --git a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterDeserializer.kt b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterDeserializer.kt index 05081abdd..cdab3a9cd 100644 --- a/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterDeserializer.kt +++ b/quartz/src/jvmAndroid/kotlin/com/vitorpamplona/quartz/nip01Core/relay/filters/FilterDeserializer.kt @@ -21,6 +21,7 @@ package com.vitorpamplona.quartz.nip01Core.relay.filters import com.fasterxml.jackson.core.JsonParser +import com.fasterxml.jackson.core.JsonToken import com.fasterxml.jackson.databind.DeserializationContext import com.fasterxml.jackson.databind.deser.std.StdDeserializer import com.fasterxml.jackson.databind.node.ObjectNode @@ -32,11 +33,149 @@ class FilterDeserializer : StdDeserializer(Filter::class.java) { override fun deserialize( jp: JsonParser, ctxt: DeserializationContext, - ): Filter = ManualFilterDeserializer.fromJson(jp.codec.readTree(jp)) + ): Filter = ManualFilterDeserializer.fromJson(jp) } class ManualFilterDeserializer { companion object { + /** + * Streaming filter parser. Reads field-by-field off [jp] without + * materializing an intermediate `JsonNode` tree — same shape as + * [com.vitorpamplona.quartz.nip01Core.jackson.EventDeserializer], + * which is what makes high-fan-out REQ traffic cheap on the + * relay-inbound path. + * + * Caller must position the parser so [jp.currentToken] is the + * `START_OBJECT` opening the filter. On return, [jp.currentToken] + * is the matching `END_OBJECT`. + * + * Tolerant by design — invalid array entries (wrong type, JSON + * null) are silently dropped, mirroring the + * `mapNotNull { it.asTextOrNull() }` behavior of the tree-based + * overload below. Unknown top-level fields are skipped via + * [JsonParser.skipChildren]. + */ + fun fromJson(jp: JsonParser): Filter { + var ids: MutableList? = null + var authors: MutableList? = null + var kinds: MutableList? = null + var tags: MutableMap>? = null + var tagsAll: MutableMap>? = null + var since: Long? = null + var until: Long? = null + var limit: Int? = null + var search: String? = null + + while (jp.nextToken() != JsonToken.END_OBJECT) { + val name = jp.currentName() + jp.nextToken() // advance to value + when { + name == "ids" -> { + ids = readStringArray(jp) + } + + name == "authors" -> { + authors = readStringArray(jp) + } + + name == "kinds" -> { + kinds = readIntArray(jp) + } + + name == "since" -> { + if (jp.currentToken != JsonToken.VALUE_NULL) since = jp.longValue + } + + name == "until" -> { + if (jp.currentToken != JsonToken.VALUE_NULL) until = jp.longValue + } + + name == "limit" -> { + if (jp.currentToken != JsonToken.VALUE_NULL) limit = jp.intValue + } + + name == "search" -> { + if (jp.currentToken != JsonToken.VALUE_NULL) search = jp.text + } + + name.length > 1 && name[0] == '#' -> { + val map = tags ?: mutableMapOf>().also { tags = it } + map[name.substring(1)] = readStringArray(jp) + } + + name.length > 1 && name[0] == '&' -> { + val map = tagsAll ?: mutableMapOf>().also { tagsAll = it } + map[name.substring(1)] = readStringArray(jp) + } + + else -> { + jp.skipChildren() + } + } + } + + return Filter( + ids = ids, + authors = authors, + kinds = kinds, + tags = tags, + tagsAll = tagsAll, + since = since, + until = until, + limit = limit, + search = search, + ) + } + + /** + * Reads a string array off [jp]. Drops non-string entries and + * JSON nulls — matches the `mapNotNull { it.asTextOrNull() }` + * tolerance of the tree-based path. Returns an empty list when + * the value is anything other than a `START_ARRAY` (incl. + * `null`), so callers don't have to special-case that. + */ + private fun readStringArray(jp: JsonParser): MutableList { + val out = mutableListOf() + if (jp.currentToken == JsonToken.START_ARRAY) { + while (jp.nextToken() != JsonToken.END_ARRAY) { + if (jp.currentToken == JsonToken.VALUE_STRING) { + out.add(jp.text) + } else if (jp.currentToken == JsonToken.START_OBJECT || jp.currentToken == JsonToken.START_ARRAY) { + jp.skipChildren() + } + } + } else if (jp.currentToken == JsonToken.START_OBJECT) { + jp.skipChildren() + } + return out + } + + /** + * Reads an int array off [jp]. Same tolerance rules as + * [readStringArray] — non-numeric entries are dropped. + */ + private fun readIntArray(jp: JsonParser): MutableList { + val out = mutableListOf() + if (jp.currentToken == JsonToken.START_ARRAY) { + while (jp.nextToken() != JsonToken.END_ARRAY) { + when (jp.currentToken) { + JsonToken.VALUE_NUMBER_INT, JsonToken.VALUE_NUMBER_FLOAT -> out.add(jp.intValue) + JsonToken.START_OBJECT, JsonToken.START_ARRAY -> jp.skipChildren() + else -> Unit + } + } + } else if (jp.currentToken == JsonToken.START_OBJECT) { + jp.skipChildren() + } + return out + } + + /** + * Tree-based overload kept for callers that already have an + * `ObjectNode` in hand (e.g. cross-format adapters). New code on + * the relay-inbound path should use the streaming overload — + * this one materializes the full filter tree first. + */ fun fromJson(jsonObject: ObjectNode): Filter { val tagsIn = mutableListOf() jsonObject.fieldNames().forEach { diff --git a/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt b/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt index 370021e30..a5d060a3b 100644 --- a/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt +++ b/quic/interop/src/main/kotlin/com/vitorpamplona/quic/interop/runner/InteropClient.kt @@ -209,10 +209,33 @@ fun main() { // ipv6 — same flow over an IPv6 socket; // JDK DatagramChannel.connect handles // the v6 address resolution natively. + // ecn — runner verifies ECN-CE counts in + // the pcap. Client just does a 100KB + // transfer; the IP-layer ECT codepoint + // is set by the sim and we don't + // need to do anything special. + // amplificationlimit — runner verifies server obeys 3x + // amplification limit. Pure server + // check — client does a normal + // transfer (the runner sets + // TESTCASE_CLIENT=transfer). + // blackhole — sim drops ALL packets for several + // seconds mid-transfer; client must + // resume after blackhole ends. Our + // PTO + retransmit handles this; the + // runner sets TESTCASE_CLIENT=transfer. + // keyupdate — server initiates a 1-RTT key update + // mid-transfer (KEY_PHASE bit flips). + // Our RFC 9001 §6 receive-side key + // update lands the rotation; runner + // verifies the pcap shows packets + // in both phases. Server-side test + // from our perspective. "handshake", "chacha20", "transfer", "http3", "multiplexing", "transferloss", "transfercorruption", "longrtt", "goodput", "crosstraffic", "retry", "ipv6", + "ecn", "amplificationlimit", "blackhole", // NOTE: the runner does NOT have a `versionnegotiation` testcase // (its Available list excludes it). The :quic VN-handling code // (applyVersionNegotiation, FORCE_VERSION_NEGOTIATION constant) @@ -245,6 +268,28 @@ fun main() { ) } + // keyupdate: same transfer flow but the client initiates a + // RFC 9001 §6 1-RTT key update once the handshake is + // confirmed. Runner verifies pcap shows BOTH client and + // server emitting packets in phase 1 — without our side + // initiating, aioquic's plain-transfer server doesn't + // rotate spontaneously and the test fails with "Expected + // to see packets sent with key phase 1 from both client + // and server". + "keyupdate" -> { + runTransferTest( + requests = requests, + downloadsDir = downloadsDir, + cipherSuites = cipherSuites, + offeredAlpns = offeredAlpns, + initialVersion = initialVersion, + keyLogPath = keyLogPath, + qlogDir = qlogDir, + parallel = requests.split(Regex("\\s+")).count { it.isNotBlank() } > 1, + initiateKeyUpdate = true, + ) + } + // The runner reuses TESTCASE_CLIENT=multiconnect for the // handshakeloss + handshakecorruption tests (see // testcases_quic.py:746). Each URL must be fetched on a fresh @@ -294,6 +339,7 @@ private fun runTransferTest( keyLogPath: String?, qlogDir: File?, parallel: Boolean, + initiateKeyUpdate: Boolean = false, ): Int { val urls = requests @@ -407,6 +453,27 @@ private fun runTransferTest( // runs even when DEBUG=0 — this is a control-flow boundary, // not a hot-path trace. System.err.println("[boot] transfer mode: parallel=$parallel urls=${urls.size}") + + // RFC 9001 §6 keyupdate testcase: the runner verifies the pcap + // shows packets in BOTH key phases from BOTH sides. Without + // initiating from our side, only the server's natural rotation + // (if any) would show — aioquic's transfer-server doesn't + // initiate, so we'd see only phase 0. Initiate after the + // handshake is confirmed (HANDSHAKE_DONE → status=CONNECTED, + // RFC 9001 §6.5 prerequisite) but BEFORE we send the GET so the + // request itself is in phase 1 — the server's response then + // mirrors phase 1, satisfying the runner's check. Brief poll + // for status because awaitHandshake returns on TLS-done + // (1-RTT keys derived) which is one ack ahead of HANDSHAKE_DONE + // arriving. + if (initiateKeyUpdate) { + withTimeoutOrNull(2_000L) { + while (conn.status != QuicConnection.Status.CONNECTED) delay(10) + } + conn.initiateKeyUpdate() + System.err.println("[boot] keyupdate: client initiated rotation to phase 1") + } + val outcome = withTimeoutOrNull(TRANSFER_TIMEOUT_SEC * 1_000L) { val responses = diff --git a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt index e3429a19d..32d33a91d 100644 --- a/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt +++ b/quic/src/commonMain/kotlin/com/vitorpamplona/quic/connection/QuicConnection.kt @@ -1481,6 +1481,81 @@ class QuicConnection( qlogObserver.onKeyUpdated("client", EncryptionLevel.APPLICATION) } + /** + * RFC 9001 §6.1 — initiate a 1-RTT key update from our side. Derive + * next-phase send keys (and pre-derive next-phase receive keys, for + * the inevitable response from the peer) using HKDF-Expand-Label + * "quic ku", install both as live, flip the phase fields. The next + * outbound packet carries `KEY_PHASE = 1` and the peer is expected + * to mirror back in the same phase. + * + * RFC 9001 §6.5 says an endpoint MUST NOT initiate a key update + * before the handshake is confirmed (HANDSHAKE_DONE received). The + * caller is responsible for that check; this method just performs + * the rotation. §6.4 also forbids initiating a second update before + * the current one has been confirmed (peer responds in matching + * phase) — same caller contract. + * + * Header-protection key is unchanged (RFC 9001 §6.1: HP key is NOT + * rotated when keys are updated). + * + * Returns true if rotation succeeded; false if app keys aren't yet + * installed (handshake hasn't completed) or the cipher suite isn't + * cached. The interop runner's keyupdate testcase requires the + * client to send packets in phase 1 — without this method we'd + * only echo peer-initiated rotations and the test fails with + * "Expected to see packets sent with key phase 1 from both client + * and server". + */ + fun initiateKeyUpdate(): Boolean { + val cs = appCipherSuite.takeIf { it != 0 } ?: return false + val curRx = appReceiveSecret ?: return false + val curTx = appSendSecret ?: return false + val liveRx = application.receiveProtection ?: return false + val liveSend = application.sendProtection ?: return false + + // Derive next-phase secrets and protections for both directions + // up front. We MUST roll both sides because the peer responds in + // the new phase — if our receive state is still at the old phase + // when their response lands, the receive-side commit path will + // re-derive the SAME keys we just installed (idempotent but + // wasteful) and then promote them, ending up with our previous + // receive keys orphaned in [previousReceiveProtection]. + val nextRxSecret = + com.vitorpamplona.quic.crypto.HKDF + .expandLabel(curRx, "quic ku", ByteArray(0), curRx.size) + val nextTxSecret = + com.vitorpamplona.quic.crypto.HKDF + .expandLabel(curTx, "quic ku", ByteArray(0), curTx.size) + val freshRx = packetProtectionFromSecret(cs, nextRxSecret) + val freshTx = packetProtectionFromSecret(cs, nextTxSecret) + + previousReceiveProtection = liveRx + application.receiveProtection = + com.vitorpamplona.quic.connection.PacketProtection( + aead = freshRx.aead, + key = freshRx.key, + iv = freshRx.iv, + hp = liveRx.hp, + hpKey = liveRx.hpKey, + ) + application.sendProtection = + com.vitorpamplona.quic.connection.PacketProtection( + aead = freshTx.aead, + key = freshTx.key, + iv = freshTx.iv, + hp = liveSend.hp, + hpKey = liveSend.hpKey, + ) + appReceiveSecret = nextRxSecret + appSendSecret = nextTxSecret + currentReceiveKeyPhase = !currentReceiveKeyPhase + currentSendKeyPhase = !currentSendKeyPhase + qlogObserver.onKeyUpdated("client", EncryptionLevel.APPLICATION) + qlogObserver.onKeyUpdated("server", EncryptionLevel.APPLICATION) + return true + } + /** Caller must hold [lock]. Snapshot of streams for the driver's send loop. */ internal fun streamsLocked(): Map = streams