Update sent to an uncached Worker after a command-less first Workflow Task runs before the run method, then fails replay

Aperta
  • Qualcosa non va?
#2,080 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
18/100
Tipo di issue
Bug
Chiarezza
Abbastanza chiara
Stato di attività
Attiva
Stack tecnologico
python
Ambito
backend, distributed-systems

Direzione di ricerca

Start with _WorkflowInstanceImpl.activate in temporalio/worker/_workflow_instance.py (around lines 484-500 in 1.34.0), which applies update and signal jobs before initialize_workflow, and compare it with the replay path. Run the repro script with max_cached_workflows=0 and then with a large cache to confirm the ordering difference. Done means the run method always executes before the update handler on an uncached worker, and the repro completes without a nondeterminism error.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

An update that reaches a Worker without the Workflow in its cache, right after a first Workflow Task that issued no commands, runs its handler before the run method, both live. A later replay runs the run method first. So values from workflow.uuid4() swap between the two, and every later Workflow Task fails with a nondeterminism error.

temporalio 1.34.0, Python 3.13.16, Temporal CLI 1.10.0 (Server 1.32.2), macOS.

import asyncio
import uuid
from datetime import timedelta

from temporalio import activity, workflow
from temporalio.api.enums.v1 import EventType
from temporalio.client import WorkflowUpdateStage
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker


@activity.defn
async def noop() -> None:
    pass


@workflow.defn
class UpdateWorkflow:
    def __init__(self) -> None:
        self.done = False

    @workflow.run
    async def run(self) -> None:
        mode = "replay" if workflow.unsafe.is_replaying() else "live"
        print(f"run method ({mode}):", workflow.uuid4(), flush=True)
        await workflow.wait_condition(lambda: self.done)

    @workflow.update
    async def update(self) -> None:
        mode = "replay" if workflow.unsafe.is_replaying() else "live"
        activity_id = f"act-{workflow.uuid4()}"
        print(f"update ({mode}):", activity_id, flush=True)
        await workflow.execute_activity(
            noop, activity_id=activity_id, start_to_close_timeout=timedelta(seconds=10)
        )
        self.done = True


async def main() -> None:
    async with await WorkflowEnvironment.start_local(dev_server_log_level="error") as env:
        async with Worker(
            env.client,
            task_queue="q",
            workflows=[UpdateWorkflow],
            activities=[noop],
            max_cached_workflows=0,
        ):
            handle = await env.client.start_workflow(
                UpdateWorkflow.run, id=str(uuid.uuid4()), task_queue="q"
            )
            # Wait until the first Workflow Task, which has no commands, completes.
            while not any(
                e.event_type == EventType.EVENT_TYPE_WORKFLOW_TASK_COMPLETED
                for e in (await handle.fetch_history()).events
            ):
                await asyncio.sleep(0.1)
            await handle.start_update(
                UpdateWorkflow.update, wait_for_stage=WorkflowUpdateStage.ACCEPTED
            )
            try:
                await asyncio.wait_for(handle.result(), 10)
                print("completed")
            except asyncio.TimeoutError:
                for e in (await handle.fetch_history()).events:
                    if e.event_type == EventType.EVENT_TYPE_WORKFLOW_TASK_FAILED:
                        print("task failed:", e.workflow_task_failed_event_attributes.failure.message)
                        break
                await handle.terminate()


if __name__ == "__main__":
    asyncio.run(main())

Output of python repro.py 2>/dev/null | grep -E '^(run method|update|task failed|completed)':

run method (live): d2101c6b-5154-42d0-9d93-90790dce80d8
update (live): act-d2101c6b-5154-42d0-9d93-90790dce80d8
run method (live): fe143ce3-789d-406b-97bb-ee4e6ddb64ab
run method (replay): d2101c6b-5154-42d0-9d93-90790dce80d8
update (replay): act-fe143ce3-789d-406b-97bb-ee4e6ddb64ab
run method (replay): d2101c6b-5154-42d0-9d93-90790dce80d8
update (replay): act-fe143ce3-789d-406b-97bb-ee4e6ddb64ab
task failed: [TMPRL1100] Nondeterminism error: Activity id of scheduled event 'act-d2101c6b-5154-42d0-9d93-90790dce80d8' does not match activity id of activity command 'act-fe143ce3-789d-406b-97bb-ee4e6ddb64ab'
  • Line 1: the first Workflow Task runs the run method, which draws d2101c6b-… and waits. It issues no commands.
  • Lines 2 and 3: the update's task, on a cache of 0, runs the update handler first, live, so it draws d2101c6b-…, then runs the run method again, also live.
  • From line 4: each later task replays the run method first, so the handler draws fe143ce3-… and its Activity ID no longer matches history.

With max_cached_workflows=1000, the same script prints:

run method (live): 12225546-2e7a-4fbc-8a12-dea82961db18
update (live): act-70f81baf-de7d-47d4-ba6c-8ffa093f1a15
completed

_WorkflowInstanceImpl.activate (temporalio/worker/_workflow_instance.py, lines 484-500 in 1.34.0) applies signal and update jobs before initialize_workflow. That suits signal-with-start, but here the run method already ran in an earlier task. It may be that the first task should be replayed on its own instead.

Found while reviewing temporalio/ai-integrations#133, where the LangSmith plugin's run IDs depend on the same order.

Lingua principale
Python
Stelle
1.2k
Fork
285
Merge medio
1g 23h
PR unite (30g)
89

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di temporalio/sdk-python

Tutte le issue di temporalio/sdk-python

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.