[Bug] Permanent worker deadlock: any CustomSlotSupplier + heartbeating activities (GIL ↔ sdk-core mutex ABBA)

[odedkedemdreeze]

What are you really trying to do?

Run a worker with a custom slot supplier (temporalio.worker.CustomSlotSupplier) that gates activity admission based on available memory, where the activities heartbeat. Nothing exotic — the supplier's callback bodies are irrelevant to this bug (a no-op supplier reproduces it).

Describe the bug

Any worker that uses any CustomSlotSupplier and runs activities that heartbeat will eventually deadlock permanently — in our production workload (heartbeat-dense, bursty activity starts) roughly 29 times per 24h across the fleet, and in the minimal reproducer below within ~1–2 seconds. The asyncio event loop and a tokio worker thread wedge against each other; pollers stop, the event loop stops, and the process emits no further logs (even unrelated Python threads freeze, since none can acquire the GIL). Recovery requires killing the process — a zombie worker.

It is a classic ABBA deadlock between the Python GIL and sdk-core's outstanding_activity_tasks parking_lot::Mutex. It is structural — it lives in the FFI calling convention, not in user code:

Thread B — asyncio event loop (holds GIL, wants mutex)
activity.heartbeat() → temporalio/worker/_activity.py → bridge record_activity_heartbeat (temporalio/bridge/worker.py:258 → temporalio/bridge/src/worker.rs:657–666 at current main a557803). That pyo3 fn is synchronous with no py.allow_threads (there is no allow_threads anywhere in bridge/src/worker.rs), so it holds the GIL for the whole call while core's record_heartbeat blocks on outstanding_activity_tasks.lock() (sdk-core crates/sdk-core/src/worker/activities.rs:447 at core main 6f75de5).

Thread A — tokio worker, activity task start (holds mutex, wants GIL)
The PendingStart branch (crates/sdk-core/src/worker/activities.rs:597–605) runs:

self.outstanding_tasks.lock().insert(
    tt.clone(),
    RemoteInFlightActInfo::new(&task.resp, task.permit.into_used(ActivitySlotInfo {...})),
);

Per Rust temporary-lifetime rules the mutex guard is alive while the insert arguments evaluate, so into_used → the supplier's mark_slot_used runs under the mutex — and the Python bridge's mark_slot_used wrapper (temporalio/bridge/src/worker.rs:427–445) begins with an unconditional Python::attach, which blocks waiting for the GIL.

Thread A: holds outstanding_activity_tasks mutex ── wants GIL (mark_slot_used → Python::attach)
Thread B: holds GIL (heartbeat FFI, no allow_threads) ── wants outstanding_activity_tasks mutex

Neither can proceed. Permanent.

Note that the supplier's callback body is irrelevant: the deadlock is the unconditional GIL acquisition before the body runs, invoked while the mutex is held. No CustomSlotSupplier implementation can avoid it.

Minimal Reproduction

Save the file below as deadlock_repro.py, then:

uv run deadlock_repro.py          # PEP 723 header pins temporalio==1.30.0
# or: pip install "temporalio==1.30.0" && python deadlock_repro.py

WorkflowEnvironment.start_local() downloads the Temporal dev server on first run (needs network). No other setup.

deadlock_repro.py (single file, self-contained)

"""Minimal reproducer: any CustomSlotSupplier + activity heartbeats permanently
deadlocks the Temporal Python worker (GIL <-> sdk-core mutex ABBA).

No application logic — a no-op slot supplier and an activity that heartbeats in a
tight loop are enough. The wedge is structural, in the SDK's calling convention,
not in anything the supplier or activity does.

The two threads that form the cycle
-----------------------------------
Thread B (asyncio event loop): activity.heartbeat() -> bridge
    record_activity_heartbeat, a synchronous pyo3 fn with NO py.allow_threads ->
    HOLDS THE GIL while sdk-core record_heartbeat() blocks on the
    outstanding_activity_tasks parking_lot::Mutex.

Thread A (tokio worker): a new activity task start evaluates
    outstanding_tasks.lock().insert(tt, RemoteInFlightActInfo::new(.., permit.into_used(..)))
    -- the mutex guard is alive while the insert arguments evaluate (Rust temporary
    lifetimes) -- into_used -> the custom supplier's mark_slot_used -> Python::attach
    -> BLOCKS waiting for the GIL.

A holds the mutex and wants the GIL; B holds the GIL and wants the mutex.
Permanent deadlock. Pollers die, the event loop dies, the process emits no more
logs -- a "zombie worker."

Why mark_slot_used being a no-op does not matter
------------------------------------------------
The deadlock is the bridge's unconditional Python::attach *before* the callback
body runs. No implementation of a CustomSlotSupplier (in any language) can avoid
it -- the bug is that a lang callback is invoked at all while outstanding_tasks
is held.

Verified against temporalio 1.30.0 (pinned sdk-core commit 9f83b7e); the same
lock ordering is byte-identical back to 1.28.0 (pinned core 6a8355a) and still
present on both repos' main as of 2026-07-14.

Run
---
    uv run deadlock_repro.py

Expected: "completed=" progress climbs for a few seconds, then STOPS. Within
DUMP_AFTER_SECONDS faulthandler prints every thread's stack; the asyncio thread's
top frame is record_activity_heartbeat, blocked in the core mutex acquisition.
The process never exits on its own -- kill it with Ctrl-C.
"""

# /// script
# requires-python = ">=3.10"
# dependencies = ["temporalio==1.30.0"]
# ///

from __future__ import annotations

import asyncio
import faulthandler
import threading
import time
from datetime import timedelta

from temporalio import activity, workflow
from temporalio.common import RetryPolicy
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import (
    CustomSlotSupplier,
    FixedSizeSlotSupplier,
    SlotMarkUsedContext,
    SlotPermit,
    SlotReleaseContext,
    SlotReserveContext,
    Worker,
    WorkerTuner,
)

TASK_QUEUE = "slot-supplier-deadlock"
CONCURRENT_ACTIVITIES = 12
TOTAL_ACTIVITIES = 8000
# Activities scheduled but not yet finished, held below Temporal's 2000
# per-workflow pending-activity limit. A window well above the slot cap keeps a
# steady stream of task-starts churning against the heartbeat FFI.
PENDING_WINDOW = 200
HEARTBEATS_PER_ACTIVITY = 150
DUMP_AFTER_SECONDS = 20

_completed = 0


class NoopCustomSlotSupplier(CustomSlotSupplier):
    """A CustomSlotSupplier with a no-op mark_slot_used.

    reserve/release keep a simple concurrency cap so activity starts churn
    steadily; mark_slot_used does nothing on purpose -- the point of this
    reproducer is that an empty callback still deadlocks, because the bridge
    acquires the GIL before the body runs.
    """

    def __init__(self, max_slots: int) -> None:
        self._max_slots = max_slots
        self._used = 0
        self._lock = threading.Lock()

    async def reserve_slot(self, ctx: SlotReserveContext) -> SlotPermit:
        while True:
            permit = self.try_reserve_slot(ctx)
            if permit is not None:
                return permit
            await asyncio.sleep(0.005)

    def try_reserve_slot(self, ctx: SlotReserveContext) -> SlotPermit | None:
        with self._lock:
            if self._used >= self._max_slots:
                return None
            self._used += 1
            return SlotPermit()

    def mark_slot_used(self, ctx: SlotMarkUsedContext) -> None:
        return None

    def release_slot(self, ctx: SlotReleaseContext) -> None:
        with self._lock:
            self._used = max(0, self._used - 1)


@activity.defn
async def heartbeating_activity(beats: int) -> None:
    for i in range(beats):
        activity.heartbeat(i)
        await asyncio.sleep(0)
    global _completed
    _completed += 1


@workflow.defn
class FanOutWorkflow:
    @workflow.run
    async def run(self, total: int, beats: int, window: int) -> None:
        in_flight = asyncio.Semaphore(window)

        async def run_one() -> None:
            async with in_flight:
                await workflow.execute_activity(
                    heartbeating_activity,
                    beats,
                    start_to_close_timeout=timedelta(minutes=5),
                    heartbeat_timeout=timedelta(seconds=30),
                    retry_policy=RetryPolicy(maximum_attempts=1),
                )

        await asyncio.gather(*(run_one() for _ in range(total)))


def build_tuner(max_activities: int) -> WorkerTuner:
    fixed = FixedSizeSlotSupplier(100)
    return WorkerTuner.create_composite(
        workflow_supplier=fixed,
        activity_supplier=NoopCustomSlotSupplier(max_activities),
        local_activity_supplier=fixed,
        nexus_supplier=fixed,
    )


def start_progress_thread() -> None:
    def _loop() -> None:
        start = time.monotonic()
        last = -1
        while True:
            time.sleep(0.5)  # releases the GIL; re-acquiring it wedges once frozen
            now = _completed
            elapsed = time.monotonic() - start
            stalled = " <-- NO PROGRESS (likely wedged)" if now == last else ""
            print(f"[{elapsed:6.1f}s] completed={now}{stalled}", flush=True)
            last = now

    threading.Thread(target=_loop, name="progress", daemon=True).start()


async def main() -> None:
    faulthandler.enable()
    print(
        f"starting: {TOTAL_ACTIVITIES} activities, cap={CONCURRENT_ACTIVITIES}, "
        f"{HEARTBEATS_PER_ACTIVITY} heartbeats each",
        flush=True,
    )
    async with await WorkflowEnvironment.start_local() as env:
        async with Worker(
            env.client,
            task_queue=TASK_QUEUE,
            workflows=[FanOutWorkflow],
            activities=[heartbeating_activity],
            tuner=build_tuner(CONCURRENT_ACTIVITIES),
        ):
            start_progress_thread()
            faulthandler.dump_traceback_later(DUMP_AFTER_SECONDS, repeat=True)
            await env.client.execute_workflow(
                FanOutWorkflow.run,
                args=[TOTAL_ACTIVITIES, HEARTBEATS_PER_ACTIVITY, PENDING_WINDOW],
                id="slot-supplier-deadlock-wf",
                task_queue=TASK_QUEUE,
            )
    print("workflow COMPLETED without a deadlock -- raise the load and retry")


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

It is a no-op CustomSlotSupplier + activities that heartbeat in a loop + concurrent activity starts, against WorkflowEnvironment.start_local(). It wedges within ~1–2 seconds (usually before the first activity completes), then a faulthandler timeout dump fires at 20s showing the event-loop thread's top frame:

Thread 0x... (most recent call first):
  <no Python frame>                          # tokio worker: blocked in native Rust
                                             # (mark_slot_used -> Python::attach, wants GIL)

Thread 0x... (most recent call first):
  File ".../deadlock_repro.py", line 160 in _loop   # unrelated progress thread — also frozen,
                                                     # can't get the GIL to print

Thread 0x... (most recent call first):
  File ".../temporalio/bridge/worker.py", line 258 in record_activity_heartbeat  # <-- holds GIL, wants mutex
  File ".../temporalio/worker/_activity.py", line 290 in _heartbeat_async
  File ".../asyncio/events.py", line 89 in _run
  ...

(The tokio threads show <no Python frame> because they are blocked in native Rust; faulthandler only prints Python frames. The frozen unrelated thread is why wedged workers go completely silent — a plain Python-thread watchdog can't even dump this.)

Suggested fix

Release the GIL around the core call in record_activity_heartbeat: decode the proto first (it reads borrowed Python bytes, needs the GIL — the current code already does this first), then wrap the worker.record_activity_heartbeat(...) call in py.allow_threads(...). With the GIL released during the mutex wait, the cycle cannot form. We'll submit a PR with this change + a regression test shortly and link it here.

For completeness / the maintainers' consideration (we are not pursuing this half ourselves): the other side of the cycle is core's hold-while-calling-back pattern — hoisting RemoteInFlightActInfo::new(...) into a local before .lock() would fix it for every language SDK, and the same guard-alive-during-into_used shape also appears in core's nexus task start (worker/nexus.rs:447–462) and local-activity dispatch (worker/activities/local_activities.rs:504–513), plus a drop-order variant in LA complete() where the slot permit (whose drop runs release_slot) is released while the LA manager lock is held. Workflow-permit release paths may warrant the same audit. Either side alone breaks this particular cycle; this issue + our PR address the Python side.

Environment/Versions

• OS and processor: reproduced on macOS arm64 and Linux x86_64; mechanism (GIL + parking_lot mutex) is platform-independent
• Temporal Version: temporalio==1.30.0 (pinned core 9f83b7e); lock ordering byte-identical back to 1.28.0 (core 6a8355a); both repos' main unfixed as of 2026-07-14 (sdk-python a557803, sdk-core 6f75de5) — no released version fixes it
• Python 3.13; client-side only (server irrelevant; reproduces against the local dev server)

Additional context

• Also reported via Temporal support ticket #33912 (reproducer attached there as well).
• Production impact before we worked around it: permanent zombie workers requiring kills; the only mitigation is abandoning CustomSlotSupplier entirely.