Idempotentność to za mało: trzy błędy w pipeline’ach z ponawianiem

Opublikowano: 3 września 2026

Problem: zielone testy, czerwony pipeline

Zbudowałeś pipeline z trwałym stanem. Każdy etap ma stabilną tożsamość, w bazie obowiązuje ograniczenie unikalności, a mechanizm ponawiania obsługuje wyjątki. Testy jednostkowe przechodzą. Testy integracyjne również. Dopiero na produkcji okazuje się, że ponowne wykonanie niedeterministycznego etapu korzystającego z modelu AI generuje inny artefakt niż pierwsza próba — a kolejne etapy po cichu zaczynają pracować na niewłaściwej wartości. Innym razem równoległe zadanie odczytuje nieaktualny stan, ponieważ wykonało się przed zadaniem, od którego faktycznie zależało. Albo proces ulega awarii między zapisaniem artefaktu a trwałym utworzeniem następnego zadania i cały pipeline pozostaje zablokowany.

To nie są wyłącznie teoretyczne przypadki brzegowe. Takie błędy potrafią pozostać niewykryte mimo całkowicie zielonego zestawu testów, ponieważ typowe testy nie odtwarzają dokładnych przeplotów wykonania, do których dochodzi podczas ponowień. Wspólna przyczyna jest zawsze podobna: pojedynczy zapis do bazy danych jest idempotentny, ale cały proces już nie.

W tym artykule przejdziemy przez trzy różne mechanizmy awarii. Każdy z nich narusza inny niezmiennik i wymaga nieco innego testu regresyjnego. Na końcu zbierzemy te zasady w przenośny zestaw niezmienników i wzorców testowych, które pozwalają wykrywać takie błędy, zanim pojawią się na produkcji.

Akt 1: niezmiennik kanonicznego artefaktu

Awaria 1: pułapka lokalnej wartości

Rozważ etap z trwałą tożsamością, który wywołuje niedeterministyczny model AI. Jego identyfikator jest stabilny, na przykład pipeline_run_123/stage_456. Przy pierwszym wykonaniu model generuje artefakt A, który zostaje zapisany za pomocą INSERT ... ON CONFLICT DO NOTHING. Transakcja zostaje zatwierdzona, więc A istnieje już trwale w bazie. Chwilę później proces ulega awarii.

Po ponownym uruchomieniu ten sam etap generuje artefakt B. Jest inny, ponieważ model AI nie jest deterministyczny. Kod próbuje zapisać B, ale ograniczenie unikalności związane z tożsamością etapu powoduje konflikt i nowy zapis zostaje pominięty. W bazie nadal znajduje się A. Na poziomie samego zapisu wszystko wygląda więc poprawnie — operacja jest idempotentna.

Problem pojawia się chwilę później. Operacja zapisu nie zwraca utrwalonej wartości, ponieważ INSERT został pominięty, a kod po ponowieniu kontynuuje wykonanie z lokalną wartością B. Kolejne walidacje i etapy dostają więc B zamiast kanonicznego A, które faktycznie znajduje się w bazie. System staje się niespójny: trwały stan mówi A, natomiast dalsze wykonanie procesu opiera się na B.

Niezmiennik: zapis musi albo skutecznie utrwalić proponowany artefakt, albo odczytać artefakt, który już został przypisany do tej samej stabilnej tożsamości. Wszystkie kolejne decyzje muszą korzystać z wartości kanonicznej zwróconej przez tę operację. Innymi słowy: dalsze wykonanie zawsze powinno opierać się na stanie, który faktycznie wygrał w trwałym magazynie danych.

Przykłady wykorzystują pseudokod przypominający Pythona, ale opisane niezmienniki nie zależą od konkretnego języka ani środowiska uruchomieniowego.

# Bug: retry używa lokalnej wartości po nieudanym insercie
def save_artifact_buggy(stage_id: str, artifact: bytes) -> bytes:
    try:
        db.execute(
            "INSERT INTO artifacts (stage_id, artifact) VALUES (%s, %s) ON CONFLICT DO NOTHING",
            (stage_id, artifact)
        )
        # Bug: brak wartości zwrotnej; caller kontynuuje z lokalnym 'artifact'
    except Exception:
        pass
    return artifact  # Zwraca lokalną wartość przejściową, nie kanoniczną

# Fix: spróbuj zapisać; jeśli ktoś już zapisał, odczytaj to, co faktycznie wygrało
def save_artifact_canonical(stage_id: str, artifact: bytes) -> bytes:
    inserted = db.fetchone(
        """
        INSERT INTO artifacts (stage_id, artifact)
        VALUES (%s, %s)
        ON CONFLICT (stage_id) DO NOTHING
        RETURNING artifact
        """,
        (stage_id, artifact)
    )
    if inserted:
        return inserted[0]  # Wygraliśmy insert; nasza wartość jest kanoniczna
    # Artefakt kogoś innego już posiada tę tożsamość stage'a: odczytaj go
    row = db.fetchone(
        "SELECT artifact FROM artifacts WHERE stage_id = %s",
        (stage_id,)
    )
    return row[0]  # Kanoniczny, utrwalony artefakt — nigdy lokalna wartość przejściowa

Awaria 2: pułapka kolejności równoległych zadań

Rozważ teraz przejście w workflow, które tworzy dwa zadania: A aktualizuje zewnętrzny system na podstawie kanonicznego wyniku, a B przygotowuje następny etap. W implementacji kryje się niejawne założenie, że B zobaczy skutki działania A. Scheduler nie zna jednak tej zależności. Traktuje A i B jako dwa niezależne zadania gotowe do wykonania, więc nic nie gwarantuje ich kolejności.

Możliwy przebieg wygląda następująco:

Zmiana stanu workflow zostaje zatwierdzona.
Worker pobiera zadanie B.
B odczytuje stan, zanim pojawią się efekty wykonania A.
B wybiera niewłaściwą gałąź, kończy się błędem albo przygotowuje nieaktualne dane wejściowe.
Dopiero później A wykonuje się poprawnie.

Ponawianie zadań ani idempotentność ich obsługi nie naprawiają naruszonej kolejności. Cały pipeline może nawet ostatecznie dojść do końca, ale po drodze zdąży już podjąć decyzję na podstawie nieprawidłowego stanu.

Sama kolejność dodania zadań do kolejki nie tworzy relacji happens-before. To najważniejsza zasada w tym przypadku. Niezmiennik: jeżeli zadanie B wymaga efektów działania A, zależność A → B musi zostać zapisana w trwałym stanie workflow jako rzeczywista krawędź happens-before. Alternatywnie zmiana stanu wymagana przez B oraz utworzenie samego B muszą zostać powiązane transakcyjnie.

# Bug: siblingy z niejawną kolejnością
def emit_jobs_buggy(stage_id: str):
    # Oba joby emitowane jako niezależne siblingy
    workflow.emit_job("update_external", stage_id)
    workflow.emit_job("prepare_next", stage_id)  # Brak gwarancji kolejności

# Fix: łańcuch zależnego joba z pomyślnego poprzednika
def emit_jobs_chained(stage_id: str):
    # A emituje B dopiero po swoim pomyślnym zakończeniu
    workflow.emit_job("update_external", stage_id, on_success=["prepare_next"])

Awaria 3: pułapka okna między zapisem a awarią

Rozważ pipeline, w którym etap AI generuje wersję roboczą, zapisuje ją jako artefakt kanoniczny, a następnie tworzy zadanie odpowiedzialne za jej weryfikację. Przebieg może wyglądać tak:

Etap AI generuje wersję roboczą D.
D zostaje trwale zapisana jako wersja kanoniczna.
Proces ulega awarii, zanim zadanie weryfikacyjne zostanie trwale utworzone.
Po ponowieniu etap stwierdza, że D już istnieje.
Idempotentny zapis niczego nie zmienia.
Ponieważ utworzenie zadania weryfikacyjnego następowało wyłącznie po udanym, pierwotnym INSERT, weryfikacja nie zostaje już nigdy zaplanowana.

Rezultat jest szczególnie podstępny: kanoniczny artefakt istnieje, ale proces nie ma możliwości przejścia do kolejnego kroku. Stan jest trwały, poprawny lokalnie i jednocześnie niekompletny. Pipeline utknął.

Warto zwrócić uwagę na rolę wzorca outbox w tym scenariuszu. Nie chodzi tu przede wszystkim o niezawodne przesyłanie komunikatów, lecz o atomowe zapisanie dwóch rzeczy: zmiany stanu oraz zamiaru wykonania dalszego kroku. Niezmiennik: utrwalenie artefaktu oraz utworzenie — albo zagwarantowanie istnienia — jego dalszej kontynuacji muszą zostać powiązane transakcyjnie. Jeśli nie można tego zrobić w jednej transakcji, system musi mieć mechanizm uzgadniania, który potrafi wykryć i odtworzyć brakującą kontynuację.

# Bug: utrwalenie i kontynuacja nie są atomowe
def process_draft_buggy(stage_id: str, draft: bytes):
    # Utrwal draft
    db.execute(
        "INSERT INTO drafts (stage_id, draft) VALUES (%s, %s) ON CONFLICT DO NOTHING",
        (stage_id, draft)
    )
    # Zaplanuj weryfikację (może zawieść po utrwaleniu, ale przed commitem)
    workflow.emit_job("verify_draft", stage_id)

# Fix: transactional outbox paruje utrwalenie i kontynuację
def process_draft_atomic(stage_id: str, draft: bytes):
    with db.transaction():
        # Utrwal draft
        db.execute(
            "INSERT INTO drafts (stage_id, draft) VALUES (%s, %s) ON CONFLICT DO NOTHING",
            (stage_id, draft)
        )
        # Zapisz event kontynuacji do tabeli outbox
        db.execute(
            "INSERT INTO outbox (event_type, payload) VALUES ('verify_draft', %s)",
            (json.dumps({"stage_id": stage_id}),)
        )
    # Publisher outboxu wyemituje job weryfikacji po commicie

Wzorce testów regresyjnych dla każdej awarii

Najważniejsza zasada stojąca za tymi testami jest prosta: sprawdzaj zachowanie systemu na granicach odtwarzania po awarii, a nie wyłącznie końcowy rezultat biznesowy. Właśnie w tych punktach ujawniają się różnice między „operacja jest idempotentna” a „cały proces jest bezpieczny przy ponowieniu”.

Test pułapki lokalnej wartości — niezmiennik kanonicznego artefaktu

def test_canonical_artifact_on_retry():
    """
    Weryfikuje, że retry po crashu używa kanonicznego, utrwalonego artefaktu,
    nie lokalnej wartości przejściowej.
    """
    stage_id = "test_run/stage_1"
    # Pierwsze wykonanie zapisuje artefakt A
    artifact_a = b"artifact A"
    save_artifact_canonical(stage_id, artifact_a)

    # Retry (po crashu) wylicza inny artefakt B
    artifact_b = b"artifact B"
    result = save_artifact_canonical(stage_id, artifact_b)

    # Downstream musi zobaczyć A, nie B
    assert result == artifact_a

Test pułapki kolejności równoległych zadań

def test_sibling_ordering():
    """
    Weryfikuje, że zależny job nie wykona się przed swoim wymaganiem.
    """
    # Workflow z dwoma siblingami, gdzie B zależy od efektów A
    workflow = Workflow()
    workflow.add_job("A", effects=["write_state_x"])
    workflow.add_job("B", depends_on=["A"], reads=["state_x"])

    # Uruchom workflow ze schedulerem, który może przeplatać
    # Ten test powinien zawieść, jeśli B wykona się przed zakończeniem A
    result = workflow.run()
    assert result["B"]["state_x"] == result["A"]["state_x"]

Test pułapki okna między zapisem a awarią — niezmiennik transakcyjnej kontynuacji

def test_crash_window_between_persist_and_continuation():
    """
    Crash na każdej granicy między utrwaleniem, insertem do outboxu, commitem i claimem.
    Weryfikuje, że ostatecznie istnieje dokładnie jeden kanoniczny draft
    i weryfikacja staje się wykonywalna.
    """
    for crash_point in ["after_persist", "after_outbox_insert", "after_commit", "after_claim"]:
        # Reset stanu
        db.execute("DELETE FROM drafts")
        db.execute("DELETE FROM outbox")

        # Uruchom pipeline z crashem w wskazanym punkcie
        try:
            run_pipeline_with_crash(crash_point)
        except SimulatedCrash:
            pass

        # Retry pipeline'u
        run_pipeline()

        # Weryfikacja: istnieje dokładnie jeden kanoniczny draft
        drafts = db.fetchall("SELECT * FROM drafts")
        assert len(drafts) == 1

        # Weryfikacja: job weryfikacji jest wykonywalny
        outbox_events = db.fetchall("SELECT * FROM outbox WHERE event_type = 'verify_draft'")
        assert len(outbox_events) == 1

Podsumowanie: trzy niezmienniki bezpiecznego ponawiania

Niezmiennik kanonicznego artefaktu: jedna trwała tożsamość etapu musi odpowiadać dokładnie jednemu kanonicznemu, utrwalonemu artefaktowi. Każde ponowienie musi kontynuować pracę na tym właśnie artefakcie, a nie na lokalnej wartości powstałej podczas kolejnej próby. Samo ograniczenie unikalności w bazie jest konieczne, ale niewystarczające. Trzeba jeszcze zagwarantować, że wartość, która faktycznie wygrała zapis, jest również wartością przekazywaną dalej przez pipeline.

Niezmiennik trwałej zależności: jeżeli zadanie B wymaga efektów zadania A, ta zależność musi istnieć w trwałym stanie systemu. Powinna mieć postać jawnej krawędzi happens-before w grafie zadań workflow, a nie wynikać jedynie z kolejności, w jakiej oba zadania zostały dodane do kolejki.

Niezmiennik atomowej kontynuacji: zatwierdzenie zmiany stanu i zapis zamiaru wykonania następnego kroku muszą być atomowe. W praktyce oznacza to zwykle rekord zadania albo wpis w tabeli outbox zapisany w tej samej transakcji. Jest to inna własność niż trwała zależność: zależność określa kolejność między zadaniami, które już istnieją, natomiast atomowa kontynuacja zapobiega sytuacji, w której zatwierdzony artefakt zostaje bez żadnego dalszego kroku.

Te trzy niezmienniki, połączone z odpowiednimi testami regresyjnymi, pozwalają wykrywać błędy związane z ponawianiem, które bez problemu przechodzą przez zwykłe zestawy testów. Najważniejszy wniosek jest prosty: idempotentny zapis w bazie danych nie oznacza jeszcze idempotentnego procesu. Trzeba zagwarantować, że każda kolejna próba zbiega do tego samego kanonicznego stanu i kontynuuje pracę właśnie od niego — a nie tylko trafia w ten sam wiersz tabeli.

Komentarze (0)

Brak komentarzy.

Dodaj komentarz

Komentarze są publikowane po moderacji. Adres e-mail nie będzie widoczny publicznie.