Trzy bugi retry safety, które przeżywają zielone testy w niedeterministycznych pipeline'ach AI

Problem: zielone testy, czerwony pipeline

Zbudowałeś trwały (durable) pipeline. Każdy stage ma stabilną tożsamość, constraint unikalności na wierszu bazy danych i handler retry, który łapie wyjątki. Testy jednostkowe przechodzą. Testy integracyjne przechodzą. A potem w produkcji retry niedeterministycznego stage'a AI produkuje inny artefakt niż pierwsza próba — i kolejne stage'y po cichu użyją złej wartości. Albo sibling job czyta nieaktualny stan, bo wykonał się przed swoją zależnością. Albo crash między zapisem artefaktu a zakolejkowaniem następnego joba zostawia pipeline trwale zablokowany.

To nie są hipotetyczne przypadki brzegowe. To konkretne tryby awarii, które przeżywają zielone suite'y testowe, bo testy nigdy nie odtwarzają dokładnych przeplotów (interleavings), jakie występują przy retry. Root cause jest w każdym przypadku ten sam: zapis do bazy danych jest idempotentny, ale pipeline jako całość — nie.

Ten artykuł przeprowadza przez trzy różne mechanizmy awarii, każdy z własnym invariantem i kształtem testu regresyjnego. Na końcu dostaniesz przenośny zestaw invariantów i wzorców testowych, które łapią te bugi, zanim trafią do produkcji.

Akt 1: Invariant kanonicznego artefaktu

Awarie 1: pułapka lokalnej wartości

Rozważ trwały stage, który woła niedeterministyczny model AI. Stage ma stabilną tożsamość (np. pipeline_run_123/stage_456). Pierwsza próba wylicza artefakt A i zapisuje go przez INSERT ... ON CONFLICT DO NOTHING. Wiersz jest zacommitowany. Potem proces się wywala.

Przy retry stage wylicza artefakt B (inny, bo model AI jest niedeterministyczny). Retry próbuje zapisać B, ale constraint unikalności na tożsamości stage'a odrzuca insert. Wiersz w bazie pozostaje A. Dotąd wszystko dobrze — baza jest idempotentna.

Ale tu jest bug: operacja zapisu nie zwraca żadnej utrwalonej wartości (bo insert został zignorowany), a retry kontynuuje, używając swojej lokalnej, przejściowej wartości B. Dalsze gate'y i kolejne stage'y widzą B, nie kanoniczne A, które zostało zacommitowane. Pipeline jest teraz niespójny: baza mówi A, a stan pipeline'u mówi B.

**Invariant:** operacja zapisu musi albo utrwalić proponowany artefakt, albo odczytać artefakt, który już posiada stabilną tożsamość. Każda dalsza decyzja musi używać zwróconej wartości kanonicznej. To jest invariant *canonical-first execution*.

Przykłady używają pseudokodu w stylu Pythona; same invarianty są niezależne od runtime'u i języka.

# 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

Awarie 2: pułapka kolejności siblingów

Teraz rozważ transition workflow'u, który emituje dwa joby: A (aktualizacja zewnętrznego systemu wynikiem kanonicznym) i B (przygotowanie następnego stage'u). Niejawne założenie: B obserwuje efekty wyprodukowane przez A. Ale scheduler traktuje A i B jako niezależne, wykonywalne siblingy — nie istnieje między nimi żadna relacja kolejności.

Oto zaobserwowany przeplot:

1. Transition się commituje.

2. B zostaje zabrany (claimed) przez workera.

3. B czyta stan, zanim efekty A istnieją.

4. B wybiera nieprawidłową gałąź, zawodzi albo przygotowuje nieaktualne dane wejściowe.

5. A wykonuje się pomyślnie dopiero potem.

Retry i idempotentność handlerów nie naprawiają tego naruszenia kolejności. Pipeline może ostatecznie się zakończyć, ale z nieprawidłowym stanem pośrednim.

Kolejność enqueue nie tworzy relacji happens-before — to jest najważniejszy jednozdaniowy takeaway tej awarii. **Invariant:** jeśli zależny job wymaga efektów sibling joba, relacja A-potem-B musi istnieć jako trwała (durable) krawędź happens-before w stanie workflow'u, albo wymagana mutacja stanu i utworzenie B muszą być sparowane 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"])

Awarie 3: pułapka okna crashu

Rozważ pipeline, który generuje draft przez stage AI, utrwala go jako kanoniczny draft, a potem kolejkuje job weryfikacji. Timeline:

1. Stage AI generuje draft D.

2. D zostaje utrwalony jako kanoniczny draft.

3. Proces się wywala, zanim job weryfikacji zostanie trwale utworzony.

4. Przy retry stage widzi, że D już istnieje.

5. Naiwny idempotentny zapis nie robi nic.

6. Ponieważ utworzenie joba było sprzężone tylko z oryginalną, pomyślną ścieżką insertu, weryfikacja nigdy nie zostaje zaplanowana.

Wynik: kanoniczny artefakt istnieje, ale lifecycle jest trwale niekompletny. Pipeline utknął.

Zwróć uwagę, czym jest tu outbox: nie chodzi przede wszystkim o reliable messaging — chodzi o atomowe zapisanie zmiany stanu i zamiaru kontynuacji. **Invariant:** utrwalenie artefaktu i utworzenie (lub istnienie) jego zależnej kontynuacji muszą być sparowane transakcyjnie albo odzyskiwalnie pogodzone (reconciled).

# 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

Kształty testów regresyjnych dla każdej awarii

Zasada stojąca za tymi testami: testuj granice odzyskiwania (recovery boundaries), nie tylko biznesowe rezultaty.

Test pułapki lokalnej wartości (invariant 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 siblingów

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 crashu (invariant 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 invarianty dla retry-safe pipeline'ów

1. **Invariant kanonicznego artefaktu:** identyczna, trwała tożsamość stage'a implikuje dokładnie jeden kanoniczny, utrwalony artefakt. Wszystkie retry muszą kontynuować z tego artefaktu, nie z lokalnej wartości przejściowej. Unikalność wiersza jest konieczna, ale niewystarczająca — chyba że utrwalony zwycięzca jest jednocześnie wartością propagowaną przez pipeline.

1. **Invariant trwałej zależności (durable dependency):** jeśli zależny job wymaga efektów sibling joba, relacja musi istnieć w trwałym stanie — krawędź happens-before w grafie jobów workflow'u, a nie jedynie kolejność, w jakiej joby zostały zakolejkowane.

1. **Invariant atomowej kontynuacji:** commit zmiany stanu i zapis zamiaru wykonania następnego kroku muszą być atomowe — typowo rekord joba albo wiersz outboxu zapisany w tej samej transakcji. To jest inna własność niż invariant zależności: jeden dotyczy kolejności między istniejącymi jobami, drugi — nigdy nie zostawiania zacommitowanego artefaktu bez jego kontynuacji.

Te trzy invarianty, w połączeniu z powyższymi kształtami testów regresyjnych, wyłapią bugi retry safety, które przeżywają zielone suite'y testowe. Kluczowy wniosek: idempotentność bazy danych to nie idempotentność pipeline'u — musisz zapewnić, że każdy retry konwerguje na tym samym kanonicznym stanie, a nie tylko na tym samym wierszu bazy danych.