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

Offen
  • Stimmt etwas nicht?
#2,080 0 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen

Maintainer antworten meist innerhalb von 1 Tag

Dieses Issue hat noch niemand übernommen.

Bewertung

Schwierigkeit
4/5
Geschätzter Aufwand
3-5 Tage
Anfängerfreundlichkeit
18/100
Issue-Typ
Bug
Klarheit
Größtenteils klar
Aktivitätsstatus
Aktiv
Tech-Stack
python
Bereich
backend, distributed-systems

Rechercherichtung

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.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Beschreibung

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.

Vorherrschende Sprache
Python
Sterne
1.2k
Forks
286
Ø Merge
1 T. 23 Std.
Gemergte PRs (30 T.)
80

Entwicklungsumgebung

Erste Schritte

  1. Lesen Sie das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreiben Sie ins Issue, dass Sie es übernehmen — das erspart doppelte Arbeit.
  3. Forken Sie das Repository und arbeiten Sie in einem Branch.
  4. Öffnen Sie einen Pull Request, der die Issue-Nummer nennt.

Mehr aus temporalio/sdk-python

Alle Issues in temporalio/sdk-python

Ähnliche Issues

Weitere Issues zu Python

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.