Po początkowym zachwycie korzyściami systemów rozproszonych szybko można dostrzec związane z nimi wyzwania. Jeden z Twitterowiczów wskazał konkretnie dwa:

There are only two hard problems in distributed systems: 2. Exactly-once delivery 1. Guaranteed order of messages 2. Exactly-once delivery

W tym artykule skupimy się na problemie gwarantowanej kolejności przetwarzania wiadomości. Nie jest to typowy poradnik, nie będziemy przechodzić od problemu do rozwiązania krok po kroku. Skupię się na różnych zagadnieniach związanych z kolejnością wiadomości, pokażę potencjalne scenariusze i sposoby, które stosowałem w praktyce.

Artykuł nie jest jednak „silver bullet”, który pozwoli całkowicie wyeliminować ten problem w Twoim projekcie. Nie bez powodu kolejność wiadomości jest uznawana za jedno z większych wyzwań w asynchronicznych systemach rozproszonych. Nie chcę również rozpisywać się o specyficznych rozwiązaniach wykorzystujących technologie, z którymi nie mam doświadczenia (Kafka, RabbitMQ). Na co dzień pracuję z rozwiązaniami AWS-owymi i w kontekście pracy z nimi powstał ten artykuł.

Na czym polega problem?

Sam problem jest stosunkowo prosty do zrozumienia. Spodziewamy się otrzymania co najmniej dwóch wiadomości w określonej kolejności. Jeśli wiadomości dotrą w pożądanej przez nas kolejności, wtedy wszystko działa zgodnie z oczekiwaniami. Jeśli jednak wiadomości przyjdą w innej kolejności, wtedy potencjalnie możemy spodziewać się problemów. Mówiąc o kolejności wiadomości, należy omówić rodzaje uporządkowania:

• Total (global) ordering – zachowanie kolejności wiadomości na poziomie globalnym. Zwykle nie jest wymagany i w praktyce jest trudny we wdrożeniu i utrzymaniu;
• Per-entity ordering – zachowanie relacji czasowej pomiędzy zdarzeniami dotyczącymi tego samego bytu, np. użytkownika, koszyka, subskrypcji czy pliku. Nie interesuje nas zachowanie kolejności globalnie. Mając zdarzenia A1, A2, B1, B2, interesuje nas relacja A1 względem A2 i B1 względem B2. Kombinacje A1, A2, B1, B2, A1, B1, A2, B2, B1, A1, A2, B2 będą poprawne. Jednocześnie np. kombinacja A1, A2, B2, B1 będzie niepoprawna. W dalszej części artykułu przez „kolejność” będę rozumiał właśnie ordering per-entity.

Należy również odróżnić kolejność dostarczenia wiadomości od kolejności ich przetwarzania i zapisu stanu. Nawet jeśli broker zachowuje kolejność wiadomości FIFO, aplikacja może ją naruszyć, np. poprzez równoległe przetwarzanie kilku wiadomości. Dlatego myśląc o kolejności, warto patrzeć na cały proces, od opublikowania wiadomości, przez wykonanie logiki, aż do zacommitowania zmian w bazach danych.

Analizując procesy biznesowe, łatwo je rozpatrywać w sposób sekwencyjny, jak na poniższym przykładzie.

Taki sposób rozumienia jest dla nas prosty i intuicyjny. Spotykamy się z nim na co dzień. Większość czynności, które robimy w życiu, to procesy liniowe.

Tworząc kod, zwykle organizujemy go w zbiór warunków i wywołań metod wykonywanych linia po linii. Co więcej, przy pierwszej styczności z algorytmami, modelowaniem procesów pokazywane są w formie listy kroków czy liniowych diagramów, co dodatkowo utrwala liniowy sposób myślenia. Ten sposób myślenia jest w pełni poprawny, gdy mówimy o systemach rozproszonych działających w sposób synchroniczny. W tym przypadku problem kolejności wiadomości jest zwykle łatwiejszy do kontrolowania, ale nadal może występować.

W przypadku systemów rozproszonych asynchronicznych warto spojrzeć na proces nie jako na listę kroków, lecz jako na zestaw zdarzeń, które w nim zachodzą.

Z punktu widzenia samego procesu nic się nie zmieniło. By proces przebiegł prawidłowo, zdarzenia muszą przyjść w określonej kolejności. Nie możemy np. opłacić zamówienia przed jego stworzeniem. Jednak myślenie o procesie w kontekście zdarzeń pomaga szybciej dostrzec takie zależności i ewentualne scenariusze, gdy mogą wystąpić.

Warto pamiętać też, że procesy w systemie mogą dziać się równolegle, często na tych samych zasobach. Może to wynikać zarówno z tego, że z aplikacji korzysta w tym samym czasie wielu użytkowników, jak i z aspektów wydajnościowych, gdzie sekwencyjne uruchamianie pewnych procesów nie ma racji bytu ze względu na czas wykonania procesu.

Dlaczego ten problem występuje?

Mówiąc o systemach rozproszonych, należy wspomnieć o Fallacies of Distributed Computing. To zbiór twierdzeń opracowanych przez L. Petera Deutscha i innych z Sun Microsystems, opisujących fałszywe założenia przyjmowane przez programistów pracujących z aplikacjami rozproszonymi:

1. The network is reliable;
2. Latency is zero;
3. Bandwidth is infinite;
4. The network is secure;
5. Topology doesn’t change;
6. There is one administrator;
7. Transport cost is zero;
8. The network is homogeneous;

Patrząc na wskazane założenia, można wskazać co najmniej kilka aspektów mogących sprawić, że ostatecznie wiadomości nie dotrą do odbiorcy lub odbiorców w oczekiwanej kolejności.

Niezawodność sieci nie jest gwarantowana. Założenie, że sieć zawsze działa i nie ma opóźnień, jest dość naiwne. Na sieć mogą wpływać zarówno czynniki zależne od nas, jak i niezależne. Po naszej stronie problem może wynikać z błędów konfiguracji, niedostatecznej liczby zasobów czy zwiększonego ruchu. Istnieje również wiele czynników niezależnych od nas, takich jak problemy po stronie dostawcy usług, na której stoi nasza usługa, mniejsze lub większe awarie czy w skrajnych przypadkach zdarzenia takie jak pożar serwerowni. W zdecydowanej większości czasu problemów nie będzie, ale złośliwość rzeczy martwych niestety się zdarza, szczególnie w przypadku głośnych premier czy innych czarnych piątków.

Część z opisanych czynników może odpowiadać za opóźnienia, a te zaś mogą być na tyle duże, że w konsekwencji zdarzenia przyjdą w nieoczekiwanej kolejności.

Problem może też wynikać z przyjętej decyzji architektonicznej, np. wyboru topologii messagingu. Jednym z wyborów jest topologia topic-per-type. W tej topologii każdy typ wiadomości w systemie ma dedykowany topic, a każdy odbiorca ma dedykowaną kolejkę do odbioru wiadomości. Każdy message opublikowany w danym topicu jest replikowany n razy, gdzie n to liczba powiązanych kolejek, a następnie jest umieszczany na kolejkach.

W konsekwencji dla każdego odbiorcy powstaje tyle kolejek, na ile wiadomości nasłuchuje. Kolejki nadal są strukturami FIFO, jednak jedynie w obrębie danej kolejki.

W przedstawionej sytuacji tracimy kontrolę nad tym, z której kolejki consumer zdejmie pierwszą wiadomość. Realnym scenariuszem jest sytuacja, w której aplikacja wykorzystująca omawianą topologię staje się nagle popularna i każdy chce założyć w niej konto. Jednocześnie otrzymujemy bardzo dużo wiadomości AccountCreated. Czas ich przetwarzania się przedłuża, przez co SubscriptionCreated może być przetwarzane, zanim AccountCreated zostanie zdjęte z kolejki.

Czy to oznacza, że topologia topic per type jest zła? Niekoniecznie, po prostu jest wrażliwa na ten rodzaj problemu. Co więcej, problemu złej kolejności wiadomości można również doświadczyć, jeśli ją zmodyfikujemy. Ograniczając do minimum liczbę topiców i kolejek, możemy skończyć z topologią, w której wszystkie wiadomości lądują na jednym topicu, a ten wrzuca je na jedną kolejkę.

Problem wielu kolejek znika, więc powinniśmy mieć zachowane FIFO. Jednak FIFO nie oznacza, że kolejne wiadomości na kolejce czekają na przetworzenie poprzednich. Odbiorca może ściągnąć kilka wiadomości w tym samym czasie i przetwarzać je równolegle. Jeśli wiadomości muszą być przetworzone w określonej kolejności, to nie mamy żadnej gwarancji co do kolejności ich przetwarzania. Przetwarzanie wiadomości pojedynczo zwykle nie jest dobrym pomysłem, chociażby ze względów wydajnościowych. A nawet jeśli byśmy je przetwarzali pojedynczo, to w przypadku dostawienia kolejnych instancji aplikacji problem wróci – każda będzie miała swój osobny limit przetwarzania. Dochodzi tu do zjawiska competing consumers, gdzie mamy kilku konsumentów wiadomości z danej kolejki i nie mamy kontroli nad tym, kto i co zdejmie ze wskazanej kolejki.

„U mnie ten problem nie wystąpi”

Konsekwencje tego podejścia najlepiej zobrazuje przykład. Nasz system to typowa aplikacja SaaS w modelu subskrypcyjnym. Klient najpierw musi utworzyć konto, a następnie wybrać subskrypcję. Dla uproszczenia załóżmy, że wymaga to wykonania dwóch zapytań do backendu – utworzenia konta oraz utworzenia subskrypcji. Cały proces przedstawia poniższy diagram.

Wysłanie zapytania na POST:/accounts powoduje rozpoczęcie asynchronicznego procesu tworzenia konta i w odpowiedzi zwraca identyfikator, którym będzie reprezentowany użytkownik. Analogicznie wygląda tworzenie subskrypcji. W tym przypadku wymagane jest wskazanie powiązanego użytkownika. W przypadku, gdy wskazany użytkownik nie istnieje, system zwraca błąd 404.

Pojawia się dość oczywiste pytanie – co w przypadku, gdy proces tworzenia konta jeszcze się nie zakończył? Załóżmy, że przekazaliśmy do QA zadanie, by sprawdzić, czy taki przypadek jest możliwy. W trakcie testów okazało się, że proces utworzenia konta zajmuje kilka sekund. Dodatkowo proces tworzenia subskrypcji zajmuje na tyle długo, że nie udaje się odtworzyć takiego przypadku. Idąc z duchem reguł KISS i YAGNI, moglibyśmy dojść do wniosku, że „działa to nie ruszaj” będzie tu dobrym podejściem.

W tym podejściu widzę co najmniej dwa problemy. Pierwszy wiąże się z Fallacies of Distributed Computing – szczególnie istotne są dwa pierwsze założenia. Wszelkie opóźnienia czy awarie mogą wpłynąć na kolejność przetwarzania wiadomości.

Drugim problemem jest zmienność wymagań systemowych. Załóżmy, że przyszło wymaganie, by uprościć proces tworzenia subskrypcji. Utworzenie konta subskrypcji ma dziać się w ramach jednego formularza i wymagać tylko jednego kliknięcia. Czas między zapytaniami drastycznie się skraca. Ryzyko wystąpienia sytuacji, gdy konto użytkownika jeszcze nie istnieje przy tworzeniu subskrypcji, istotnie wzrasta.

Mam nadzieję, że zgadzamy się, że problem istnieje i może wystąpić. Jednak to, czy trzeba coś z tym robić, to już jest bardzo indywidualna kwestia. Każdy scenariusz, w którym utworzenie subskrypcji skutkuje błędem, to potencjalnie stracony klient. Każdy musi sobie sam odpowiedzieć, czy przymknięcie oka na istnienie tego problemu i ewentualnych szkód w konkretnym przypadku jest racjonalne. Pewnie znajdą się przypadki, gdy zignorowanie tego problemu będzie bardziej opłacalne niż jego łatanie.

Projektując systemy komunikujące się w sposób asynchroniczny, przydatną heurystyką jest „zakładanie najgorszego” – sieć będzie wolna, serwery wysypią się w najgorszym momencie, zdarzenia będą zduplikowane lub przyjdą w złej kolejności. Cytując Vlada Khononova (Learning Domain-Driven Design: Aligning Software Architecture and Business Strategy):

The word driven in event-driven architecture means your whole system depends on successful delivery of the messages. Hence, avoid the “things will be okay” mindset like the plague.

Ponieważ celem tego artykułu jest omówienie problematyki kolejności wiadomości, pomijam również wariant przeniesienia rozwiązania na podejście synchroniczne. Być może komunikacja asynchroniczna nie jest stworzona do tego przypadku? Jeśli dane podejście nie jest adekwatne do klasy problemu, nie ma co go używać na siłę.

Retry to the rescue

Jednym z klasycznych przypadków problemów wynikających z błędnej kolejności przetwarzania jest próba akcji na nieistniejącym zasobie. Wiadomość tworząca zasób albo jeszcze nie dotarła, albo jeszcze jest przetwarzana. Najprostsze, co można wtedy zrobić, to poczekać na nią i po jakimś czasie ponowić próbę przetworzenia wiadomości.

async handle( message: UserUpdated ): Promise<void> {
    await retry(
      async () => {
        const user = await this.userRepository.findById( message.userId );

        if ( !user ) {
          throw new UserNotFoundError( message.userId );
        }

        user.update( message );
        await this.userRepository.save( user );
      },
      MAX_RETRIES,
      INITIAL_DELAY_MS
    );
  }

Gdy message UserCreated jeszcze nie został przetworzony, callback przekazany do retry wrzuci błędem, a logika retry sprawi, że po określonym czasie próba zostanie ponowiona, aż do osiągnięcia sukcesu lub do liczby prób określonej w MAX_RETRIES.

W logice retry by zwiększyć stabilność systemu można skorzystać ze strategii exponential backoff, czyli wykładniczego wzrostu czasu między kolejnymi próbami. Czasami stosuje się też Fibonacci backoff, oparty na ciągu Fibonacciego.

W przypadku przeciążenia systemu wykorzystanie backoffu pomoże w niepogorszeniu sytuacji. Problem może tu też np. wynikać z locka założonego na rekordzie usera w bazie. Zwiększenie czasu między próbami zwiększa szansę, że lock do kolejnej próby zostanie zdjęty. Dodatkową pomocą w podanych sytuacjach może być implementacja jittera, czyli losowego przesunięcia w czasie między kolejnymi próbami. W przypadku wielu powtarzających się retryów pomaga to rozłożyć ruch w czasie.

Drugim i moim zdaniem dużo lepszym, podejściem do retry jest zrobienie tego na poziomie infrastruktury. Ciężko mi wyobrazić sobie system, gdzie każde przetwarzanie wiadomości wyglądałoby w ten sposób, jak na pokazanym kodzie.

Przetwarzanie na poziomie infrastruktury przebiega w analogiczny sposób jak na poziomie kodu aplikacyjnego. Handler rzuca wyjątek, wiadomość wraca na kolejkę, serwis ponownie pobiera wiadomość. Po n nieudanych próbach wiadomość trafia na Dead Letter Queue (DLQ), czyli specjalną kolejkę, dla wiadomości, których nie udało się przetworzyć. W tym podejściu również jest możliwe wykorzystanie exponential backoff. Szczegóły implementacyjne zależeć będą jednak od wykorzystanego brokera. Np. w przypadku AWS SQS może to wyglądać jak na poniższym diagramie.

Mechanizm backoff można w przypadku SQS zaimplementować poprzez usunięcie i ponowne umieszczenie wiadomości na kolejce ze zwiększonym DelaySeconds w ramach kolejnych prób.

Jeśli wszystkie próby się nie powiodą, to w przeciwieństwie do rozwiązania w kodzie nie tracimy wiadomości. Wyląduje ona na DLQ i będzie można wykonać jej redrive ręcznie. Jest to szczególnie przydatne w przypadku, gdy wiadomość nie mogła być przetworzona, np. w wyniku błędu w kodzie, lub gdy nie ma możliwości ponownego wywołania flow, które dostarczy dany message.

Retry nie jest idealnym rozwiązaniem, ale obsłuży wiele prostych przypadków, gdy kolejność wiadomości rozjeżdża się o milisekundy lub sekundy. Ostatecznie wiadomości zostaną przetworzone, choć nieco później.

Niezależnie od podejścia, handler poddany ponownemu przetworzeniu powinien być idempotentny. Wielokrotne wykonanie takiej samej operacji powinno dać takie same rezultaty jak jej pojedyncze wykonanie. W podanym przykładzie handler jest idempotentny przy założeniu, że metoda user.update() nie wprowadza żadnych efektów ubocznych poza nadpisaniem pól danymi z message. Nieidempotentne handlery mogą powodować niepożądane zmiany w danych, np. podwójne aktualizacje, inkrementacje liczników czy generowanie nowych identyfikatorów przy każdym wywołaniu. W ostateczności może to prowadzić do niespójności systemu.

Kiedy retry nie pomoże?

Przedstawiony przypadek może być w pełni obsłużony z wykorzystaniem retry. Jednak są również przypadki, w których retry nie pomoże. Załóżmy, że mamy możliwość dodania opisu konta użytkownika. Zmieniamy go, a następnie szybko poprawiamy literówkę. Jeśli message, w którym znajduje się pierwotna wartość, przyjdzie jako drugi, ostatecznie system zignoruje poprawkę użytkownika. W obu przypadkach zasób istnieje, brak pola do zastosowania retry.

Rozwiązaniem, które możemy wykorzystać do obsługi takiego przypadku, jest wersjonowanie wiadomości. W podanym przykładzie zasób w systemie otrzymałby nowe pole opisujące jego obecną wersję. Gdy system zapisuje zaktualizowane dane, aktualizuje również wersję zasobu. Jeśli zdarzenie ze starszą wersją przyjdzie później, może zostać odrzucone. Implementując wersjonowanie, należy pamiętać o tym, że wiele rekordów może być przetwarzane w tym samym czasie. Wersjonowanie musi być egzekwowane atomowo w bazie. Na poziomie zapytania aktualizującego dane w bazie należy więc sprawdzić, czy wersja nie uległa zmianie. Podejście to nazywane jest optimistic lockingiem.

Spraw, by kolejność nie miała znaczenia

Brzmi banalnie, ale takie podejście może być wykonalne. Wróćmy do przykładu z utworzeniem konta i subskrypcją. System do spójności wymaga, by do utworzenia subskrypcji istniał użytkownik. Co, gdyby pozbyć się tego wymagania? Handler obsługujący tworzenie użytkownika może sprawdzać, czy istnieją oczekujące subskrypcje. Jeśli tak, to je aktywujemy.

async handle( { userId, details }: UserCreated ): Promise<void> {
  await usersService.save( new User( userId, details ) );
  
  const subscriptions = await subscriptionsService.getPending( userId );

  for ( const sub of subscriptions ) {
    sub.activate();

    await subscriptionRepository.save( sub );
  }
}

W analogiczny sposób obsługujemy tworzenie subskrypcji. Po zapisaniu nowej subskrypcji sprawdzamy, czy istnieje przypisane jej konto użytkownika. Jeśli tak, to aktywujemy subskrypcję.

async handle( { subscriptionId, details }: SubscriptionCreated ): Promise<void> {
  const subscription = new Subscription( subscriptionId, details );
  
  const user = await usersService.findById( details.userId );
  
  if ( user ) {
    subscription.activate();
  }
  
  await subscriptionsService.save( subscription );
}

Jednak to podejście, podobnie jak retry, nie sprawdzi się wszędzie. W przypadku aktualizacji zasobu ciężko będzie uniezależnić np. wiadomość UserUpdated od message UserCreated. Nie da się zmienić np. opisu użytkownika, którego jeszcze nie ma w systemie.

Podsumowanie

Przygotowując ten artykuł, przypomniał mi się jeden z klasyków polskiego internetu.

Tym humorystycznym akcentem myślę, że warto zakończyć ten artykuł. Jestem ciekaw, czy masz jakieś ciekawe przypadki ze swojej pracy, gdzie problemem była kolejność przetwarzania wiadomości. Jeśli masz doświadczenia w walce z tym problemem, Twój komentarz będzie tu szczególnie cenny.

Zdaję sobie sprawę, że ten artykuł nie wyczerpał tematu w pełni. Nie poruszyłem np. możliwości, jakie daje wykorzystanie partycjonowania w kontekście zapewnienia kolejności przetwarzania wiadomości. Temat partycjonowania jest jednak na tyle złożony, że chciałbym mu poświęcić dedykowany artykuł.

Dopełnieniem tego artykułu będzie drugi artykuł, w którym pochylę się nad problemem exactly-once delivery. Nie bez powodu te dwa problemy są często wymieniane razem. Zachęcam, by zapisać się na mailing, by go nie przegapić.

Zachęcam również do sprawdzenia materiałów dodatkowych. Szczególnie polecam zwrócić uwagę na materiały od DevMentors, które znacznie pomogły mi w uporządkowaniu wiedzy w tym obszarze i posłużyły mi za inspirację do przygotowania tego artykułu.

Materiały dodatkowe i źródła

• Patoarchitekci (Szymon Warda, Łukasz Kałużny) – #70 Fallacies of Distributed Computing
• DevMentors – ORDERING, czyli dlaczego KOLEJNOŚĆ komunikatów nie jest oczywista?
• DevMentors – PARTYCJONOWANIE, czyli jak zapewnić kolejność przetwarzania wiadomości? | ORDERING cz. 2
• DevMentors –  Kiedy NIE UŻYWAĆ MESSAGINGU, czyli jak nie robić sobie pod górkę
• Oskar Dudycz – Ordering, Grouping and Consistency in Messaging systems
• Learning Domain-Driven Design: Aligning Software Architecture and Business Strategy – Vlad Khononov (s247)
• AWS Docs – Amazon SQS visibility timeout
• AWS Docs – Amazon SQS delay queues
• Reddit r/softwarearchitecture – Strict ordering of events
• Wikipedia – Fallacies of distributed computing
• Jeff Hodges – Notes on Distributed Systems for Young Bloods
• Projektowanie systemów rozproszonych. Wzorce i paradygmaty dla skalowalnych, niezawodnych usług z wykorzystaniem Kubernetesa

Zaobserwuj mnie w mediach społecznościowych!

LinkedIn jest głównym medium, poza blogiem, gdzie publikuję swoje przemyślenia. Zachęcam też do zaobserwowania devszczepaniak.pl na Facebooku, gdzie oprócz informacji o nowych artykułach publikuję linki do ciekawych treści.

Udostępnij wpis

Szukasz taniego i dobrego VPS? Skorzystaj z Mikrusa i przy zakupie odbierz dodatkowy miesiąc usługi za darmo!

📖 Koniecznie sprawdź ofertę księgarni Helion oferującej tysiące książek dla programistów w dobrych cenach! Kupując z tego linku, pomagasz w rozwoju bloga.

📈 Szukasz księgowości dla swojej JDG? Skorzystaj z mojego polecenia i odbierz pierwszy miesiąc za 1zł netto!

Chcesz docenić moją pracę? Postaw mi kawę!