Eviction retries forever when a deadlock interrupt lands while a finished task schedules its done callbacks

Start with _register_task and call_soon in temporalio/worker/_workflow_instance.py, then the eviction check if self._deleting and self._tasks and attempt_deadlock_interruption in temporalio/worker/_workflow.py. The repro script is in the issue and needs a temporal CLI binary. Done means eviction completes after a deadlock interrupt lands mid-callback scheduling, with a regression test that makes the finished task drop out of _tasks.

Claude (Opus 5.5):

Summary. When TMPRL1101 fires, the worker raises _InterruptDeadlockError into the workflow thread asynchronously. If that exception is delivered while a task that has just finished is scheduling its done callbacks, the self._tasks.remove callback is never scheduled. The finished task then stays in _WorkflowInstanceImpl._tasks, every eviction attempt raises RuntimeError: Eviction processed, but 1 tasks remain, and the worker retries eviction forever. Every later workflow task for that run times out until the worker process restarts. Reproduced on 1.30.0 and 1.35.0 (Python 3.14.6).

What we saw in production (1.30.0, sandboxed runner, worker under CPU starvation). A query activation that normally takes under 1 ms exceeded the 2 s deadlock budget because the pod was starved of CPU. The workflow then stayed stuck for about 85 minutes, until we deleted the pod. Its server-side workflow task timed out every 10 s in the meantime. Trimmed log, in order:

Exception in callback <_asyncio.TaskStepMethWrapper object at 0x...>()
  File "temporalio/worker/_workflow_instance.py", line 2811, in call_soon
    self._assert_not_read_only("schedule task", allow_during_delete=True)
  File "temporalio/worker/_workflow_instance.py", line 2142, in _assert_not_read_only
temporalio.worker._workflow._InterruptDeadlockError
Failed handling activation on workflow with run ID ...
temporalio.worker._workflow._DeadlockError: [TMPRL1101] Potential deadlock detected: workflow didn't yield within 2 second(s).
WARN temporalio_sdk_core::worker::workflow: Failing legacy query request ...
Failed running eviction job for run ID ..., continually retrying eviction. ...
RuntimeError: Eviction processed, but 1 tasks remain. Stack traces below:
Stack for <Task finished name='query: state (workflow: ..., run: ...)' coro=<_WorkflowInstanceImpl._apply_query_workflow.<locals>.run_query() done, ...> result=None>

Note Task finished. The query coroutine completed, but the task never left _tasks.

Mechanism (line links at 1.35.0):

1. _register_task tracks tasks with task.add_done_callback(self._tasks.remove).
2. When a task's coroutine returns, the C Task implementation schedules each done callback through the loop's call_soon, a Python method. Its first frame is an eval-breaker check.
3. On a deadlock, attempt_deadlock_interruption raises _InterruptDeadlockError in the workflow thread with _raise_in_thread. If the thread is descheduled at the start of call_soon, the exception is delivered there. It escapes the done-callback scheduling, Handle._run logs "Exception in callback", and _tasks.remove is never scheduled.
4. The eviction check if self._deleting and self._tasks now always raises, and _handle_cache_eviction retries every 2 s indefinitely. Core holds the run in the cache, so new workflow tasks for it are never processed.

Reproduction. The patch only changes timing: it sleeps 3 s at the start of call_soon when a finished query task schedules _tasks.remove. That models the thread being descheduled at that point, as happens under CPU starvation. Requires a temporal CLI (TEMPORAL_CLI=/path/to/temporal, or omit it to download one).

"""A deadlock interrupt delivered while a finished task schedules its done
callbacks leaves the task in `_WorkflowInstanceImpl._tasks`; eviction then
fails with "Eviction processed, but 1 tasks remain" forever.

The patch below models the workflow thread being descheduled (CPU starvation)
at the start of `call_soon` while a finished query task schedules
`self._tasks.remove`. It changes timing only.
"""

import asyncio
import logging
import os
import sys
import time
import uuid

import temporalio
from temporalio import workflow
from temporalio.api.enums.v1 import EventType
from temporalio.client import WorkflowQueryFailedError
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import UnsandboxedWorkflowRunner, Worker
from temporalio.worker._workflow_instance import _WorkflowInstanceImpl

STALL = {"armed": False}
_call_soon = _WorkflowInstanceImpl.call_soon


def call_soon(self, callback, *args, context=None):
    task = args[0] if args else None
    if (
        STALL["armed"]
        and getattr(callback, "__self__", None) is self._tasks
        and isinstance(task, asyncio.Task)
        and task.get_name().startswith("query:")
    ):
        STALL["armed"] = False
        time.sleep(3)  # longer than the 2 s deadlock budget
    return _call_soon(self, callback, *args, context=context)


_WorkflowInstanceImpl.call_soon = call_soon


@workflow.defn
class Idle:
    def __init__(self) -> None:
        self.pokes = 0

    @workflow.run
    async def run(self) -> None:
        await workflow.wait_condition(lambda: self.pokes > 0)

    @workflow.signal
    def poke(self) -> None:
        self.pokes += 1

    @workflow.query
    def state(self) -> int:
        return self.pokes


class Capture(logging.Handler):
    def __init__(self) -> None:
        super().__init__()
        self.messages: list[str] = []

    def emit(self, record: logging.LogRecord) -> None:
        self.messages.append(record.getMessage() + " " + str(record.exc_info[1] if record.exc_info else ""))


async def main() -> None:
    capture = Capture()
    logging.getLogger("temporalio").addHandler(capture)
    print(f"temporalio {temporalio.__version__}, Python {sys.version.split()[0]}")
    env = await WorkflowEnvironment.start_local(dev_server_existing_path=os.environ.get("TEMPORAL_CLI"))
    queue = f"repro-{uuid.uuid4()}"
    worker = Worker(env.client, task_queue=queue, workflows=[Idle], workflow_runner=UnsandboxedWorkflowRunner())
    asyncio.create_task(worker.run())
    handle = await env.client.start_workflow(Idle.run, id=f"idle-{uuid.uuid4()}", task_queue=queue)
    print("warm query:", await handle.query(Idle.state))

    STALL["armed"] = True
    try:
        await handle.query(Idle.state)
    except WorkflowQueryFailedError as error:
        print("stalled query:", error)
    await asyncio.sleep(3)
    print("eviction:", next(m.split(" Stack")[0] for m in capture.messages if "tasks remain" in m))

    await handle.signal(Idle.poke)
    await asyncio.sleep(13)  # default workflow task timeout is 10 s
    events = [event.event_type async for event in handle.fetch_history_events()]
    print("workflow task timeouts after signal:", events.count(EventType.EVENT_TYPE_WORKFLOW_TASK_TIMED_OUT))
    print("workflow completed:", EventType.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED in events)
    sys.stdout.flush()
    # The worker cannot shut down while the eviction retries.
    os._exit(0)


asyncio.run(main())

Output on 1.35.0 (1.30.0 is identical apart from the version line):

temporalio 1.35.0, Python 3.14.6
warm query: 0
stalled query: [TMPRL1101] Potential deadlock detected: workflow didn't yield within 2 second(s).
eviction: Failed running eviction job for run ID ..., continually retrying eviction. ... Eviction processed, but 1 tasks remain.
workflow task timeouts after signal: 1
workflow completed: False

Possible fix. Make the eviction check, and the task bookkeeping, independent of done-callback delivery. For example, discard finished tasks before the check:

if self._deleting:
    self._tasks = {task for task in self._tasks if not task.done()}
    if self._tasks:
        raise RuntimeError(...)

More broadly, an exception raised asynchronously can land anywhere in the instance's event-loop internals (call_soon, _run_once). This case is the one we could pin down, and other landing points may corrupt state in similar ways.

Environment. temporalio 1.30.0 in production; reproduced on 1.30.0 and 1.35.0. CPython 3.14.6. Linux x86_64 (production), macOS arm64 (repro). Temporal Server 1.31.1 (dev server for the repro).