Schalen en benchmarken van een kritieke message bus met een nieuwe indexeringsstrategie

Dit artikel is onderdeel van een serie over projecten van stagiairs in de zomer van 2026.

Aria is ons interne berichtenframework en gehoste systeem dat dagelijks meerdere terabytes aan data verwerkt. Cliënten kunnen zich abonneren op Aria om een live stream van berichten te ontvangen. Naarmate het gebruik van Aria binnen het bedrijf snel is gegroeid, moesten we mogelijkheden vinden om het systeem te optimaliseren en opnieuw te architectureren, zodat het kan schalen met de toename in datavolume en doorvoersnelheid.

Theodor Totev, een stagiair die deze zomer met het Aria-team werkte, richtte zich op het gebruik van indexering en tree-splitting om een specifiek scenario te verbeteren: hoe kunnen we het goedkoper maken voor cliënten om slechts een subset van berichten te lezen? Zijn optimalisaties leidden tot een afname van 30% in CPU-gebruik bij productie-workloads, terwijl de extreem hoge standaard van correctheid die nodig is voor een dergelijk kritiek systeem behouden bleef.

Het filteren van berichten overbelastte de servers

Wanneer Aria berichten via TCP aan een cliënt levert, wordt de recente stream — wat we de stream tip noemen — bewaard in een in-memory ringbuffer. Als een cliënt achterop raakt, kan deze recente berichten uit deze ringbuffer opvragen om weer up-to-date te komen. We noemen dit proces tip recovery.

Een uitdaging bij tip recovery is dat Aria de volledige stream van berichten opslaat, terwijl cliënten vaak alleen geïnteresseerd zijn in een veel kleiner subset van die berichten. Aria verdeelt de berichtenstroom in topics, die een hiërarchische naamruimte vormen, vergelijkbaar met een bestandssysteem. Cliënten kunnen zich abonneren op een individueel topic of op een topic subtree (alle topics onder een bepaald topic, vergelijkbaar met een globstar). Aria filtert de volledige stream tot alleen de berichten van de geabonneerde topics en levert deze aan de cliënt.

Aanvankelijk was een simpele lineaire scan voor dit filteren voldoende. Hoewel dit algoritmisch inefficiënt is, is het doorlopen van de stream vriendelijk voor de CPU-cache en daarom redelijk snel. Naarmate het aantal cliënten dat tip recovery uitvoerde echter groeide, merkten we dat onze servers worstelden met de verhoogde belasting. Sommige servers bereikten 100% CPU-benutting, waardoor cliënten "van de tip afvielen" en niet meer in staat waren om bij te komen. We voegden tijdelijk meer servers toe, maar het werd duidelijk dat we de code voor tip recovery grondig moesten herzien.

Theodor besloot dit probleem op te lossen door index-datastructuren toe te voegen die Aria kon gebruiken om berichten efficiënt uit de stream te filteren. De vraag was echter: wat moet er worden geïndexeerd? Een naïeve aanpak zou zijn om voor elk topic een index te maken, maar Aria-instanties kunnen bijna een miljoen topics hebben, wat deze aanpak onhaalbaar maakt. In plaats daarvan maakte Theodor een index voor elke topic partition. Een topic partition bestaat uit alle topics met hetzelfde prefix van twee segmenten (waarbij een segment een enkel deel van de topicnaam is, gescheiden door een slash). Bijvoorbeeld: app/codestore/commits en app/codestore/features vallen beide onder de app/codestore topic partition.

Elke topic partition krijgt een index van de locaties van zijn berichten in de Aria-stream. Wanneer een cliënt berichten opvraagt, voert Aria een n-way merge uit van de indexen met behulp van een min-heap. Hierdoor kan Aria de berichtenstroom efficiënt reconstrueren, in de juiste volgorde, voor alleen deze specifieke topic partitions.

Prototyping en benchmarking van de index

Zoals bij elk prestatiegericht werk, moesten we benchmarks uitvoeren om te valideren dat we het systeem daadwerkelijk verbeterden. We wilden ook uitgebreide tests, aangezien de logica voor het publiceren van berichten zich in het kritieke pad van Aria bevindt.

Theodor begon met het schrijven van een tool om verschillende herstelscenario's te profileren. Op die manier konden we verschillende configuraties testen, zoals het aantal topic partitions, de mate van interleaving en het aantal lezers, om te zien hoe deze variabelen de prestaties beïnvloedden.

Met de benchmark op zak implementeerde hij de eerste versie van de topic partition index. Tijdens het profileren ontdekten we dat de min-heap de grootste bottleneck in onze code was. Vroeger zouden we waarschijnlijk één implementatie kiezen die beter leek, maar tegenwoordig is het uitvoeren van experimenten goedkoop: Theodor vroeg een agent om vijf verschillende heap-implementaties op te zetten en deze gedurende de nacht te profileren. De volgende dag hadden we het antwoord: fastheapunboxed zorgde voor een prestatieverbetering van 2x in onze geïndexeerde recovery-code.

Theodor testte de nieuwe index met veel expect-tests, oefende de nieuwe logica uit in Antithesis en liet uiteindelijk een zwerm agents de code met grote nauwkeurigheid analyseren.

Gebruik van een block pool voor het opslaan van de indexen

Nadat we de index-datastructuur hadden bedacht, moesten we een efficiënte representatie vinden. We wilden geen ringbuffer voor elke index maken, omdat ringbuffers niet dynamisch in grootte kunnen veranderen. Ze zouden dan moeten worden gedimensioneerd voor het ergste scenario.

Specifiek is de indexgrootte omgekeerd proportioneel aan de berichtgrootte: kleinere berichten betekenen dat er meer berichten in de tip store passen, wat een grotere index vereist. Een Aria-bericht kan zo klein zijn als 32 bytes. Als de volledige tip store uit berichten van die grootte zou bestaan, zou de bijbehorende index 2GB zijn. Een index van 2GB per topic partition zou te veel geheugen verspillen. We wilden dus een index die dynamisch kon krimpen en groeien met het aantal berichten in de topic partition.

De uiteindelijke oplossing was het gebruik van een gedeelde pool van blocks, die elk 1024 entries bevatten. De indexen verwijzen naar een block en voegen waarden toe. Zodra alle berichten binnen een block de ringbuffer hebben verlaten, kunnen blocks uit de index worden verwijderd en hergebruikt worden voor andere indexen.

Het lezen van oude berichten duurde te lang

Theodors werk aan de indexering loste de problemen met tip recovery op, maar we hadden nog een ander probleem met de levering van berichten: cliënten die berichten van vóór de stream tip nodig hadden, moesten veel te lang wachten.

Wanneer een cliënt herstart, vraagt deze vaak alle berichten vanaf het begin van de week op voor de geabonneerde topics. We noemen dit proces initial recovery. Initiële recovery kan miljoenen, zelfs miljarden berichten omvatten. Het is essentieel dat Aria deze berichten zo snel mogelijk leest en levert. Bovendien herstarten cliënten vaak rond dezelfde tijd, waardoor het cruciaal is dat dit proces goed schaalt.

Tijdens één specifiek incident duurde een initial recovery die normaal minder dan 2,5 seconden in beslag nam, plotseling meer dan 13 minuten. Bij onderzoek bleek dat hetzelfde probleem speelde als bij de tip store: Aria las en filterde 10x meer data dan er daadwerkelijk verzonden moest worden. Het was duidelijk dat we de manier waarop berichten op schijf werden opgeslagen, moesten herzien.

Aria persisteert berichten in subtree stores

Aria slaat berichten op schijf aanvankelijk op in chronologische segmenten per topic partition. Vervolgens splitst een apart proces deze segmenten in meerdere subtree stores, die elk berichten bevatten voor een topic subtree. Deze subtree stores zijn bedoeld om een balans te vinden tussen:

• Schrijven naar veel bestanden (resulteert in efficiënt filteren, maar vereist meer werk om te combineren tot een berichtenstroom).
• Schrijven naar minder bestanden (makkelijker te combineren, maar minder efficiënt om te filteren).

Voor dit splitsen gebruikte Aria een simpele heuristiek: het koos de subtree store door de eerste drie segmenten van het topic te lezen. Als we bijvoorbeeld de topics app/options/orders/created en app/options/orders/cancelled hadden, werden deze beide geplaatst in een subtree store voor app/options/orders. Als het topic minder dan drie segmenten had, kreeg het een eigen subtree store (bijv. app/options). In feite gaf deze heuristiek elk direct kind van een topic partition een aparte subtree store.

Deze heuristiek werkte echter niet goed wanneer één topic in een store significant meer berichten had dan de andere. Als een cliënt zich alleen zou abonneren op het minder actieve topic, moest Aria alle berichten van het andere topic wegfilteren. Net als bij tip recovery zorgde dit filteren voor een enorme werklast.

Daarnaast wilden we voorkomen dat gebruikers de interne logica van Aria moesten begrijpen bij het ontwerpen van hun topic-structuur. We wilden dat gebruikers hun topics konden structureren op een manier die voor hen logisch was, in het vertrouwen dat Aria het splitsen intelligent zou afhandelen.

Theodor kreeg de taak een algoritme te ontwikkelen dat dit probleem kon voorkomen door de topic-boom intelligent te splitsen op basis van het berichtvolume per topic. Aria kan dit volume berekenen, omdat de verwerkingsstap van segmenten plaatsvindt nadat de berichtenstroom is geperverteerd.

Gegevens verzamelen en algoritmen testen

Net als bij de implementatie van de tip recovery vereiste het werk aan adaptive splitting het verkennen van verschillende opties. Theodor begon met het verzamelen van data uit diverse Aria-sessies. Vervolgens implementeerde hij verschillende algoritmen voor het splitsen en testte deze op de verzamelde data. Hij analyseerde de resultaten vanuit verschillende perspectieven:

• Hoeveel bytes moeten we negeren uit een store als we een topic lezen?
• Wat is de som van deze "verspilde bytes" over alle topics?
• Welk topic zou, indien gelezen, resulteren in de meeste verspilde bytes?
• Welk topic zou, indien gelezen, resulteren in de slechtste ratio van totale bytes ten opzichte van nuttige bytes?

Dit proces van exploratie werd ondersteund door LLM's. Theodor kon snel een web-UI "vibe-coden" die de verschillende splitsingsalgoritmen visualiseerde over echte datasets.

Het uiteindelijke algoritme

Uiteindelijk koos Theodor voor een oplossing die een greedy clustering algoritme combineerde met een binaire zoekopdracht. Dit leverde de gewenste eigenschappen op: topics met veel berichten krijgen hun eigen subtree store, terwijl kleinere topics worden samengevoegd in één enkele store.

Voorbeeld van de oude heuristiek: Een topic van 2,72GB kon in dezelfde store zitten als topics van 10MB en 28,5MB. Dit betekende dat iemand die het topic van 10MB wilde lezen, alle andere berichten moest wegfilteren. Tegelijkertijd werden kleinere topics verdeeld over drie verschillende stores, ondanks dat ze ordes van grootte kleiner waren dan de grote topics.

Met het nieuwe adaptive splitting algoritme: Het topic van 2,72GB heeft nu een eigen store, gescheiden van de topics van 28,5MB en 10MB. De kleinere topics zijn vervolgens allemaal gebundeld in één gedeelde store.

Teststrategieën voor het algoritme

Omdat adaptive splitting direct invloed heeft op hoe berichten worden opgeslagen, wilden we met grote zekerheid vaststellen dat er geen bugs waren. Het verliezen of verkeerd ordenen van een bericht zou ernstige gevolgen hebben.

1. Expect-tests: We controleerden eigenschappen van het algoritme, zoals het feit dat grote topics een eigen store kregen en kleine 'siblings' stores deelden.
2. Property-based testing: We gebruikten een bestaande test die een reeks willekeurig gegenereerde berichten op willekeurig gekozen topics invoegde en daarna het splitsingsalgoritme uitvoerde. De test las willekeurige delen van de topic-boom uit en bevestigde dat de berichten de juiste volgorde en inhoud hadden. Theodor breidde deze test uit door het splitsen van de boom te randomiseren, wat bevestigde dat de inhoud van de berichten gelijk bleef, ongeacht hoe de boom was gesplitst.
3. Antithesis: We voerden meerdere runs uit in Antithesis om te bevestigen dat de wijzigingen niets anders in Aria braken.

Resultaten van de optimalisaties

De code voor de tip indexing is in productie genomen. Voorlopige resultaten in onze staging-omgeving laten een afname van 30% in CPU-benutting zien in bepaalde real-world scenario's. De algehele latentie voor cliënten is tijdens deze periodes significant gedaald. We merkten dat servers die de indexering niet implementeerden last hadden van tail latencies van meerdere seconden, terwijl deze problemen niet voorkwamen op servers met de nieuwe code.

Adaptive splitting is uitgerold naar onze staging-omgeving en zal binnenkort in productie worden genomen.