Nahrazujeme Axon ságy stavovými event handlery: co jsme postavili místo nich

Pokračování Migrace z Axon Framework 4 na 5: co jsme se naučili. Původní článek o migraci zmiňoval, že “refaktoring ze ság na stavové handlery byl největší změnou, ale také tou nejhodnotnější.” Tenhle článek vysvětluje, jak ten refaktoring reálně vypadá — a proč jsme cestou nesáhli po EventScheduler ani DeadlineManager.

Otázka, která zrodila tenhle článek

Po vydání článku o migraci se jeden čtenář zeptal trefně:

Museli jste použít event scheduler místo deadline manageru pro stavové zpracování událostí, vzhledem k tomu, že scope deadlinů se vztahuje jen na ságy a agregáty?

Skvělá otázka — a odpověď je “ani jedno.” Nahradili jsme obojí. To je ta překvapivá část redesignu, kterou stojí za to popsat: předpoklad zabudovaný v otázce (potřebujete nějaké schedulovací primitivum z Axonu) je předpoklad, kterého jsme se vzdali. Jakmile byly ságy ze stolu, věci, které ságy skrývaly — koordinace více událostí, čekání na budoucí okamžik a orchestrace dlouhotrvajících vícekrokových workflow — se čistě rozdělily do tří nesouvisejících starostí, z nichž každá je lépe obsluhovaná nástrojem, který jsme už měli.

Tenhle článek prochází tím, co nahradilo ságy, co nahradilo deadliny, kam skončila orchestrace workflow a o jedno produkční úskalí, na které jsme cestou narazili.

Proč musely ságy pryč

Motivace pro samotný upgrade frameworku není tématem tohoto článku — ta je pokrytá v článku o migraci Axon 4 → 5. Tahle sekce je o té volbě tvaru-jako-sága, kterou jsme udělali během té migrace.

Axon 5 si ságy ponechal dosažitelné přes cestu legacy podpory. Takže striktně vzato, “museli jsme udělat redesign” je přehnané — mohli jsme existující ságy přenést více méně beze změny. Co jsme udělat nemohli, bylo považovat to za směr dopředu. Signál z frameworku byl jasný: stavové event handlery byly tam, kam měla mířit nová práce, a ságy byly na trase, kdy jsou sice podporované, ale už se do nich neinvestuje.

Ten posun byl důležitý i z druhého důvodu. Naše služba běží na plně reaktivním Spring stacku a implementace ság z v4, kterou jsme s sebou táhli, se nikdy čistě nekomponovala s reaktivním event handlingem. Blokující vstupní body se nám pořád protlačovaly zpátky do reaktivních řetězců; obcházeli jsme to, ale každá sága, kterou jsme se v4 dotkli, byla malou přítěží v kódu.

Čas migrace tu přítěž proměnil v rozhodnutí. S frameworkem signalizujícím odklon od ság a s reaktivní kompozicí už beztak známou bolestí jsme redesign vzali jako příležitost, ne jako břemeno. Jakmile jsme začali přepisovat saga use-casy ručně, všimli jsme si, že schovávaly tři nesouvisející tvary v jedné abstrakci — rozdělení, o kterém je zbytek tohoto článku.

Koordinační polovina: Stavové event handlery

První věc, kterou ságy pro nás dělaly, byla koordinace: sledovat rozpracovaný vícekrokový proces napříč několika událostmi, rozhodnout, kdy je hotový, reagovat na to dokončení. Tohle jsme nahradili obyčejným event handlerem, který čte a zapisuje řádek v Postgres tabulce.

Tabulka je generická — jeden řádek na rozpracovaný job, bez ohledu na to, který modul job vlastní:

CREATE TABLE job_state (
    id              TEXT PRIMARY KEY,
    job_type        TEXT NOT NULL,
    correlation_id  TEXT NOT NULL,
    owner_id        TEXT NOT NULL,
    total_items     INT  NOT NULL,
    completed_items INT  NOT NULL DEFAULT 0,
    failed_items    INT  NOT NULL DEFAULT 0,
    status          TEXT NOT NULL,
    metadata        JSONB,
    created_at      TIMESTAMPTZ NOT NULL,
    updated_at      TIMESTAMPTZ NOT NULL,
    completed_at    TIMESTAMPTZ,
    version         BIGINT NOT NULL DEFAULT 0,
    UNIQUE (job_type, correlation_id)
);

Tabulku obaluje malá reaktivní služba — createJob(...), markItemCompleted(...), markItemFailed(...), getJob(...), deleteJob(...). Pak jakýkoli event handler modulu, který potřebuje sága-like chování, ji prostě použije.

Konkrétní příklad — koordinace multi-batch import jobu:

@Component
@RequiredArgsConstructor
public class OrderImportEventHandler {

    private static final String JOB_TYPE = "ORDER_IMPORT";

    private final ImportWorkerService importWorker;
    private final JobStateService jobStateService;

    @EventHandler
    public Mono<Void> on(OrderImportRequested event) {
        return jobStateService
            .createJob(JOB_TYPE, event.importId(), event.requestedBy(), event.batchCount())
            .doOnSuccess(job -> importWorker.processAsync(event.importId()))
            .then();
    }

    @EventHandler
    public Mono<Void> on(BatchProcessed event) {
        return jobStateService
            .markItemCompleted(JOB_TYPE, event.importId())
            .then();
    }

    @EventHandler
    public Mono<Void> on(OrderImportCompleted event) {
        return jobStateService.deleteJob(JOB_TYPE, event.importId());
    }
}

Co ságy dělaly implicitně — “najdi mi instanci ságy pro tenhle importId, změň její stav, rozhodni, jestli má skončit” — se proměňuje ve tři obyčejná volání metod. Korelace, kterou ságy schovávají za @SagaEventHandler(associationProperty = "importId"), je teď explicitní argument. Ukecanější; zato víc debuggovatelné.

Pár vlastností, které z toho návrhu vypadnou:

• Obyčejné Spring beany. Žádný saga lifecycle, žádné @StartSaga / @EndSaga, žádné s tím spojené lifecycle podivnosti. Handler je testovatelný stejnými Spock fixturami jako jakákoli jiná reaktivní služba.
• DB-nativní introspekce. SELECT * FROM job_state WHERE status = 'IN_PROGRESS' AND created_at < now() - interval '1 hour' najde zaseknuté joby bez jakéhokoli framework-specifického nástroje.
• Přehrávatelné očekávaným způsobem. Stav jobu je side-effect projekcí událostí, stejně jako jakýkoli jiný read model. Smažte tabulku, přehrajte event store, dostanete stejný stav zpátky.
• Idempotentní. createJob před vložením kontroluje existující řádek. markItemCompleted je jedno SQL. Opětovné doručení události stav neporuší.

Deadlinová polovina: Plánované dotazy

Druhá věc, kterou ságy pro nás dělaly, byla čekání — “za N hodin, pokud X nenastalo, udělej Y.” Ve v4 tohle vyjádříte přes DeadlineManager zevnitř ságy. Když ságy zmizí, kam tohle patří?

Dvě pozorování to udělala jednoduchým:

1. “Deadline” reálně nikdy nezávisí na in-memory stavu ságy. Vždycky závisí na sloupci v databázi — kdy tenhle job vyprší, kdy se tahle naplánovaná akce stane splatnou, kdy se tenhle resource stal způsobilým pro další stav. Stav je už v databázi.
2. Doručení deadlinu v přesný okamžik je málokdy reálný požadavek. “Někdy po T, ne později než T + N minut” je téměř vždy dost.

S ohledem na tohle není správné primitivum event scheduler — je to periodický job, který se ptá databáze, co je splatné:

@Component
@RequiredArgsConstructor
public class StuckImportSweeper {

    private final ImportRepository imports;
    private final CommandGateway commandGateway;

    @Scheduled(cron = "${app.import.sweeper.cron}")
    @SchedulerLock(name = "import_stuck_sweeper")
    public void sweep() {
        imports.findStuckImports()
            .flatMap(stuck -> commandGateway
                .send(new TimeoutImportCommand(stuck.id()))
                .onErrorResume(e -> Mono.empty()))
            .then()
            .block(); // @Scheduled nečeká na reaktivní návratovou hodnotu
    }
}

To je vše. ShedLock udržuje běh singletonu napříč pody. Dotaz rozhoduje, co je splatné. Command odvádí práci. Není tu žádná sága, kterou by bylo třeba probudit, žádný deadline k registraci, žádný upcaster pro plánované události, který je třeba udržovat napříč migracemi.

Tenhle pattern absorboval všechny dřívější deadline use-casy, které jsme měli: vypršování zaseknutých operací, přechody resourcí mezi stavy po uplynutí okna, posílání periodických notifikací odběratelům. Žádný z nich nepotřeboval sub-minutovou přesnost; všechny potřebovaly trvanlivost přes restart podu. Plánovaný DB dotaz vám dá oboje.

Když je workload workflowem

Něco z toho, co žilo v našich v4 ságách, vlastně vůbec nebylo tvaru ságy — bylo tvaru workflow. Dlouhotrvající, vícekrokové, s retry, paralelními větvemi a časovači měřenými v minutách až hodinách. Tlačit tohle do ságy, nebo později do řádku v JobState, bylo vždycky neohrabané. To napětí se ukázalo všude tam, kde jsme chtěli explicitní retry politiky per krok, fan-out / fan-in přes heterogenní aktivity, nebo trvanlivé pauzy uprostřed letu, které přežijí restart procesu.

Pro tyhle sahám po Temporalu. Ne jako po náhradě ságy — jako po správném nástroji pro jiný tvar problému. Temporal vlastní workflow stav a replay; my vlastníme aktivity, které delegují na Axon command dispatch pokaždé, když změny stavu mají být event-sourced. Hranice je čistá: Temporal orchestruje, Axon zaznamenává fakta, ani jeden nesahá do stavu toho druhého.

Kompletní rozhodovací strom, kde jsme skončili:

Tenhle článek je o prvních dvou — o use-casech ve tvaru ságy. Temporal polovina je samostatný příběh; jediný důvod, proč je na téhle stránce vůbec, je že tahá hranici toho, kde stavové event handlery přestávají být správnou odpovědí. Pokud workflow potřebuje explicitní retry politiky a trvanlivé časovače per krok, nechcete si to znovu stavět nad @EventHandler a Postgres tabulkou.

Produkční gotcha: Atomické incrementy vyhrávají nad optimistickým zamykáním

“Zřejmá” implementace markItemCompleted je:

public Mono<JobState> markItemCompleted(String jobType, String correlationId) {
    return repo.findByJobTypeAndCorrelationId(jobType, correlationId)
        .flatMap(job -> {
            job.setCompletedItems(job.getCompletedItems() + 1);
            return repo.save(job);
        });
}

Tohle je read-modify-write a funguje to přesně tak dlouho, dokud události chodí jedna po druhé. První moment, kdy máte paralelní handlery pro události se stejným correlation ID — řekněme fan-out completion events z batch workera — optimistická version kontrola ze Spring Dat odmítne všechny soubežné zápisy kromě jednoho. Poražení retryují, throughput se zhroutí a mód selhání je tichý, pokud nesledujete počty OptimisticLockingFailureException.

Oprava je posunout increment dolů do databáze:

@Modifying
@Query("""
    UPDATE job_state
    SET completed_items = completed_items + 1,
        version         = version + 1,
        updated_at      = NOW(),
        status = CASE
            WHEN completed_items + 1 + failed_items >= total_items
                  AND failed_items > 0
                THEN 'COMPLETED_WITH_ERRORS'
            WHEN completed_items + 1 + failed_items >= total_items
                THEN 'COMPLETED'
            ELSE status
        END,
        completed_at = CASE
            WHEN completed_items + 1 + failed_items >= total_items THEN NOW()
            ELSE completed_at
        END
    WHERE job_type = :jobType AND correlation_id = :correlationId
    """)
Mono<Long> atomicIncrementCompleted(String jobType, String correlationId);

Jedno UPDATE udělá increment, navýšení verze i přechod stavu — atomicky, v jednom round-tripu, bez okna pro read-modify-write. Trade-off: completion logika teď žije na dvou místech (SQL CASE a jakékoli in-memory kontroly). Pro takhle jednoduché stavové přechody je to v pořádku; pro bohatší přechody byste celé to přesunuli do stored procedury, nebo přijali cenu optimistického zamykání a retryovali.

Tohle je typ detailu, který by před námi v4 saga abstrakce schovala. Není to očividně regrese, že teď píšeme SQL ručně. Kontenční pattern tam byl pořád; framework ho jen obalil v retry smyčce.

Jak se to bude srovnávat s saga modulem v Axon 5.2

K datu Axon 5.1 jsou ságy stále na roadmapě 5.2 a nedorazily — totéž platí o DeadlineManager, EventScheduler a upcasterech. Takže srovnání níže je výhled dopředu, proti tomu, co bylo signalizováno pro 5.2, ne proti tomu, co můžeme dnes spustit. Přirozená otázka, jakmile 5.2 přistane: vytrhneme JobStateService a osvojíme si saga modul?

Pravděpodobně ne pro use-casy, které tenhle článek pokrývá, a pravděpodobně ano pro ty, které dnes nemáme.

Co bychom obětovali přechodem zpátky k ságám:

• Atomické SQL incrementy. Saga stores serializují celou ságu; vysoce kontenční koordinace by buď retryovaly, nebo potřebovaly pečlivý sharding podle saga identifikátoru.
• DB-nativní introspekci. Stav ságy žije v saga storu, ne v tabulce, kterou by si operations dashboardy mohly SELECTnout.
• Obyčejnou testovatelnost. Saga test fixtury jsou mocné, ale framework-specifické; naše jsou jen Spock proti službě.
• Čistý rozdělení mezi koordinací a čekáním. Ságy je znovu spojí.

Co by nám ságy daly zpátky:

• Méně boilerplatu. @SagaEventHandler(associationProperty = "importId") je kompaktnější než jobStateService.markItemCompleted(JOB_TYPE, importId) opakované napříč handlery.
• In-saga deadliny. scheduleDeadline(Duration.ofHours(24), "timeout") uvnitř saga logiky je přirozenější vyjádření “reaguj na tuhle událost, nebo reaguj na její absenci po prodlevě,” než dělení toho mezi handler a sweeper.
• Snapshot-based replay ság. Pro ságy, které akumulují skutečně komplexní stav (ne jen “už jsme viděli všechny batche”), je automatické snapshotování reálnou ergonomickou hodnotou.

V praxi nás téměř žádné z našich aktuálních workloadů netáhne k ságám. Tvar, který by nás lákal — workflow dost dlouhé na to, aby potřebovalo trvanlivou koordinaci, s retry a časovači — už žije v Temporalu, kde abstrakce sedí pořádně. Případy, které matchují saga sweet spot, jsou užší, než vypadají: proces dost krátký na to, aby roztočit Temporal workflow bylo overkill, ale s deadlinem závislým na akumulovaném in-memory stavu, ne jen na sloupci, který umíme sweepovat na rozvrhu. Žádný takový aktuálně nemáme. Pro workloady v tomhle článku bychom si nechali to, co máme.

Zpět k původní otázce

Je to bystrá otázka a framework historicky spojoval tyhle starosti dohromady — ságy vlastnily jak multi-event koordinaci, tak deadliny, které se odpalovaly proti akumulovanému stavu ságy. Takže očekávat one-for-one náhradu (EventScheduler místo DeadlineManager) je přirozené čtení.

Co jsme zjistili, je, že to nejsou jedna starost se dvěma API povrchy; jsou to tři starosti sdílející jeden domov ve v4. Nahradili jsme všechny tři, různými nástroji podle tvaru. Multi-event koordinace se stala řádky JobStateService. Časově založené triggery na DB-resident stavu se staly Spring @Scheduled + DB dotazy. Cokoli, co chtělo skutečný workflow engine — retry, paralelní větve, trvanlivé časovače — šlo do Temporalu. Schedulovací primitiva Axonu seděla v nepříjemném středu: příliš svázaná se stavem ságy, aby fungovala jako generický scheduler, příliš tenká na vyjádření reálných workflow.

Dekompozice těch starostí udělala každý nástroj jasnějším o tom, na co je. Takže krátká odpověď na vaši otázku: nevybírali jsme mezi event schedulerem a deadline managerem, protože ten výběr samotný byl špatný rámec.

Závěrečná myšlenka

Migrace Axon 4 → 5 nás donutila přepracovat každou ságu, kterou jsme měli. Jako u většiny vynucených redesignů ta práce nevypadala jako pokrok, dokud jsme ji dělali. Výsledkem je menší povrch problému: tři účelově padnoucí nástroje — state service nad Postgresem, @Scheduled sweeper a Temporal pro opravdové workflow — každý pokrývá jeden tvar toho, co ságy dělaly, s lepší operabilitou a žádnými framework-specifickými abstrakcemi, které je třeba učit nové inženýry.

Pokud se saga modul v 5.2 vrátí s čistým reaktivním příběhem a inline podporou deadlinů, rádi po něm sáhneme příště, až budeme modelovat proces, jehož deadline závisí na akumulovaném in-memory stavu a který je příliš krátkodobý na to, aby ospravedlnil workflow engine. Pro všechno ostatní se “správný nástroj na každý tvar” ukázal být upgradem.

Mimochodem k Axon 5.1. Tenhle článek je o sága polovině migrace. Mimo ten příběh: 5.1 vrátil agregátní snapshotování zpět, přidal first-class Spring podporu do AxonTestFixture a zavedl JSpecify nullability anotace. Stojí za vědomí, pokud jste přešli rovnou z release 5.0.x a nesledovali point release — žádné z toho nemění argument tady, ale návrat snapshotů je reálný zisk pro horké agregáty s dlouhou event historií.

Titulní foto: Eric Prouzet na Unsplash.