[BUG] `AsyncLogSender.publish` is not thread-safe and can wedge the worker event loop

[hobbyhack]

Environment

• hatchet-sdk 1.33.2
• Python 3.13.13 (CPython, Linux x86_64)

---

Summary

hatchet_sdk.worker.runner.utils.capture_logs.AsyncLogSender.publish calls self.q.put_nowait(record) on an asyncio.Queue, but publish is invoked from any thread Python's logging module dispatches into — including worker-pool threads created by asyncio.to_thread(...) inside a step.

The queue is bound to one event loop. Cross-thread put_nowait calls _wakeup_next → call_soon on the wrong loop, which under loop.set_debug(True) raises RuntimeError: Non-thread-safe operation invoked on an event loop other than the current one, and under default (debug-off) silently corrupts that loop's internal call queue. The corruption manifests as a wedged Hatchet worker that still heartbeats and receives start step run messages but never executes them ("Waiting Steps N blocked_for=Xs", "THE TIME TO START THE TASK RUN IS TOO LONG, THE EVENT LOOP MAY BE BLOCKED").

---

Stack trace (under loop.set_debug(True))

File "hatchet_sdk/worker/runner/utils/capture_logs.py", line 154, in publish
    self.q.put_nowait(record)
File "asyncio/queues.py", line 170, in put_nowait
    self._wakeup_next(self._getters)
File "asyncio/queues.py", line 76, in _wakeup_next
    waiter.set_result(None)
File "asyncio/base_events.py", line 835, in call_soon
    self._check_thread()
File "asyncio/base_events.py", line 872, in _check_thread
    raise RuntimeError(
        "Non-thread-safe operation invoked on an event loop other "
        "than the current one")
RuntimeError: Non-thread-safe operation invoked on an event loop other than the current one

---

Root cause

1. The SDK installs a LogForwardingHandler (a logging.StreamHandler subclass) on the root logger. Its emit() checks ctx_step_run_id.get() and, if set, calls self.log_sender.publish(...).
2. ctx_step_run_id is a contextvars.ContextVar. asyncio.to_thread(fn, ...) copies the current Context into the worker-pool thread, so any sync helper invoked via to_thread from within a step run inherits a populated ctx_step_run_id.
3. Common sync helpers do logging — e.g. googleapiclient calls, requests, internal logger.info(...) for observability. Each logging.info(...) from the pool thread triggers LogForwardingHandler.emit → self.log_sender.publish(record) → self.q.put_nowait(record) on a thread that is not the queue's owning loop's thread.
4. asyncio.Queue.put_nowait is documented as a single-loop (coroutine-safe) operation. It calls _wakeup_next, which calls loop.call_soon(...), which under debug raises and under release silently mutates the loop's _ready deque from a foreign thread. The race eventually leaves the loop's internal scheduling state inconsistent, and the SDK runner stops launching new step coroutines.

---

Reproduction

Any Hatchet workflow whose step delegates to a sync helper that logs.

Minimal repro:

import asyncio, logging
from hatchet_sdk import Hatchet, Context

hatchet = Hatchet()
wf = hatchet.workflow(name="repro", on_events=["repro:fire"])

@wf.task()
async def step(input, ctx: Context):
    def sync_helper():
        logging.getLogger("repro").info("hello from pool thread")
    # contextvars (incl. ctx_step_run_id) propagate into to_thread
    await asyncio.to_thread(sync_helper)
    return {}

Run with loop.set_debug(True) on the worker thread loop and fire repro:fire. The RuntimeError shown above fires immediately. Without debug mode, the same race silently corrupts loop state — fire enough events to surface the wedge (in practice, ~3 consecutive failing runs whose on-failure handlers also log from to_thread reliably wedged the worker within seconds).

---

Symptoms in production (no debug mode)

• Worker keeps heartbeating; sibling async tasks on a different loop keep running.
• Engine sends rx: start step run: <id>/<step> messages and the runner logs them, but no run: start step body or finished step run follows.
• Hatchet engine eventually emits "THE TIME TO START THE TASK RUN IS TOO LONG, THE EVENT LOOP MAY BE BLOCKED" and "Waiting Steps N" with monotonically growing blocked_for.
• Sometimes self-clears after ~15 min; often requires a worker restart.

---

Suggested fix

Capture the queue's owning loop in consume() and route publish through call_soon_threadsafe (minimal change, no new dependency):

class AsyncLogSender:
    def __init__(self, event_client):
        self.event_client = event_client
        self.q = asyncio.Queue[...](maxsize=...)
        self._owner_loop: asyncio.AbstractEventLoop | None = None

    async def consume(self) -> None:
        self._owner_loop = asyncio.get_running_loop()
        while True:
            ...

    def publish(self, record) -> None:
        loop = self._owner_loop
        if loop is None:
            return  # consume() not started yet — drop or buffer
        try:
            running = asyncio.get_running_loop()
        except RuntimeError:
            running = None
        if running is loop:
            try:
                self.q.put_nowait(record)
            except asyncio.QueueFull:
                logger.warning("log queue is full, dropping log message")
            return
        try:
            loop.call_soon_threadsafe(self._enqueue_or_drop, record)
        except RuntimeError:
            pass  # loop closed

    def _enqueue_or_drop(self, record) -> None:
        try:
            self.q.put_nowait(record)
        except asyncio.QueueFull:
            logger.warning("log queue is full, dropping log message")

---

Workaround

Monkey-patch AsyncLogSender.publish in user code along the lines of fix above (capture the loop in consume, route foreign-thread publish calls through call_soon_threadsafe).

---

Notes

Anyone who logs from inside asyncio.to_thread while a step's contextvars are set is exposed — and that's a very common pattern (HTTP clients, blocking SDKs like googleapiclient, file I/O wrapped via to_thread). The bug is silent without debug mode, so it surfaces as flaky workers in production with no obvious cause.

[mrkaye97]

hey @hobbyhack - thanks! Where are you calling loop.set_debug?

[hobbyhack]

hi @mrkaye97 no problem.

In our worker entrypoint, before asyncio.run(main()), we monkey-patch asyncio.new_event_loop so every loop the process creates — both the main async loop and the dedicated thread loop the Hatchet V1 worker spins up via worker.start() — comes up with set_debug(True) and a tuned slow_callback_duration:

import asyncio, os

def _install_asyncio_debug():
    threshold = float(os.environ.get("WORKER_ASYNCIO_SLOW_CALLBACK_S", "0.5"))
    _orig = asyncio.new_event_loop
    def _new():
        loop = _orig()
        loop.set_debug(True)
        loop.slow_callback_duration = threshold
        return loop
    asyncio.new_event_loop = _new
    os.environ.setdefault("PYTHONASYNCIODEBUG", "1")

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

The patch must run before any loop is constructed—that's why it's in the main block rather than inside main().

Enabling debug only on the main loop is not enough for this bug, because the AsyncLogSender queue lives on the Hatchet runner's loop (the one created inside worker.start() on its dedicated thread). That is the specific loop whose _check_thread() raises the exception.

To confirm without the monkey-patch, set the environment variable PYTHONASYNCIODEBUG=1 for the worker process. asyncio.new_event_loop reads this and turns on debug for every loop it creates.

The RuntimeError surfaces the moment any sync helper invoked via asyncio.to_thread(...) from inside a step calls logging.info(...) (or any logger that propagates to root).

[mrkaye97]

Gotcha, thanks! One thing I'd like a bit more context on: We don't recommend running a worker with your own event loop started - in fact, I'd be surprised if this worked at all. How does the code snippet you provided here play with when you call worker.start?

[hobbyhack]

Oops, I should have included the surrounding entrypoint.

import asyncio, threading
from hatchet_sdk import Hatchet

def _run_worker_blocking(worker):
    worker.start()                    # blocks; owns its own event loop

async def main():
    hatchet = Hatchet()
    worker = hatchet.worker("my-worker", workflows=[...])

    # Hatchet V1 worker.start() is blocking and creates its own loop,
    # so we run it in a dedicated thread alongside other async work
    # (an outbox poller) on the main loop.
    threading.Thread(
        target=_run_worker_blocking, args=(worker,),
        daemon=True, name="hatchet-worker",
    ).start()

    # ... unrelated async work on this (main) loop ...
    await asyncio.Event().wait()

if __name__ == "__main__":
    _install_asyncio_debug()           # process-wide hook (see below)
    asyncio.run(main())                # creates the main loop only

So we have two event loops in the process:

1. The main loop, created by asyncio.run(main()).
2. The Hatchet runner loop, created internally by worker.start() on the daemon thread.

We do not pass a loop to the SDK or call worker.async_start(). The SDK is in full control of its own loop.

The _install_asyncio_debug helper monkey-patches asyncio.new_event_loop before anything creates a loop. It just wraps the factory so any loop the process constructs ... main loop and the SDK's runner loop ... comes up with set_debug(True) and a tuned slow_callback_duration. Equivalent to setting PYTHONASYNCIODEBUG=1 in the environment, which I think the SDK already inherits naturally.

The bug fires inside the SDK runner loop, regardless of what the rest of the process is doing ... so it should reproduce identically with a worker-only entrypoint that just calls hatchet.worker(...).start() (no main loop, no thread).

[mrkaye97]

Gotcha, thanks! I'll look into a fix here

[hobbyhack]

Sounds great.

[mrkaye97]

Hi @hobbyhack, I looked around, but I wasn't able to reproduce this issue with the code you provided. Is there anything else I need to do to repro?

[MicroYui]

I can work on this.

I was able to reproduce the issue locally with a minimal unit-level test against the current Python SDK implementation. The repro starts AsyncLogSender.consume() on an asyncio loop with debug mode enabled, installs LogForwardingHandler, sets ctx_step_run_id, and then logs from inside asyncio.to_thread(...).

That causes the log call to go through:

LogForwardingHandler.emit -> AsyncLogSender.publish -> asyncio.Queue.put_nowait

from the worker-pool thread. Since consume() is awaiting the same queue on the event-loop thread, put_nowait() wakes the queue waiter from the wrong thread and raises:

RuntimeError: Non-thread-safe operation invoked on an event loop other than the current one

This matches the root cause described in the issue.

I’m planning to fix this by recording the queue owner loop when AsyncLogSender.consume() starts and making publish() dispatch cross-thread enqueue operations via owner_loop.call_soon_threadsafe(...). Direct enqueue can still be used when publish() is called from the owner loop. I’ll include a regression test covering logging from asyncio.to_thread(...) while Hatchet step context vars are set.

[MicroYui]

I verified the issue with a local unit-level reproduction that follows the real SDK logging path:

asyncio.to_thread(...)
  -> logging.info(...)
  -> LogForwardingHandler.emit(...)
  -> AsyncLogSender.publish(...)
  -> asyncio.Queue.put_nowait(...)
The reproduction setup was:

Start AsyncLogSender.consume() on an asyncio event loop.
Enable asyncio debug mode on that loop.
Install LogForwardingHandler.
Set ctx_step_run_id and ctx_task_retry_count.
Call logging.info(...) from inside asyncio.to_thread(...).
Because asyncio.to_thread(...) copies the active contextvars.Context, the worker-pool thread can still see ctx_step_run_id, so the log is forwarded through LogForwardingHandler.emit() into AsyncLogSender.publish().

Before this fix, publish() called self.q.put_nowait(record) directly from that worker thread. If consume() was already awaiting self.q.get() on the event-loop thread, the queue wakeup touched the owner event loop from the wrong thread. In asyncio debug mode this can surface as:

RuntimeError: Non-thread-safe operation invoked on an event loop other than the current one

On my Windows/Python 3.12 local run, the same unsafe cross-thread queue access manifested as the consumer never being woken and the test timing out while waiting for the fake event client to receive the forwarded log.

After the fix, cross-thread publishes are dispatched back to the owner event loop with owner_loop.call_soon_threadsafe(...), and the same reproduction completes successfully.

[mrkaye97]

Ah wait, is this issue only present on Windows? I know the original issue mentioned Linux - I wasn't able to repro on MacOS

[MicroYui]

I don't think this is Windows-only. The original local repro I used was on macOS, and the key is that the queue must already have a pending getter from consume() before the log is published from the worker thread.

Here is the minimal asyncio-only repro I used, without any Hatchet server dependency:

import asyncio
import contextvars
import logging
import threading
import traceback

ctx_step_run_id = contextvars.ContextVar("ctx_step_run_id", default=None)


class Sender:
    def __init__(self):
        self.q = asyncio.Queue(maxsize=10)

    async def consume(self):
        print("consume start")
        record = await self.q.get()
        print("consume done", record)

    def publish(self, record):
        print("put before from", threading.current_thread().name)
        self.q.put_nowait(record)
        print("put after")


class Handler(logging.Handler):
    def __init__(self, sender):
        super().__init__()
        self.sender = sender

    def emit(self, record):
        if ctx_step_run_id.get():
            self.sender.publish(record)


async def main():
    loop = asyncio.get_running_loop()
    loop.set_debug(True)
    print("debug", loop.get_debug())

    sender = Sender()
    handler = Handler(sender)
    logging.getLogger().addHandler(handler)
    logging.getLogger().setLevel(logging.INFO)

    task = asyncio.create_task(sender.consume())

    # Important: let consume() reach `await q.get()` so the queue has a pending getter.
    await asyncio.sleep(0.05)

    # asyncio.to_thread copies contextvars, which mirrors the SDK worker path.
    ctx_step_run_id.set("step-run-id")

    def sync_helper():
        logging.info("hi")

    try:
        await asyncio.wait_for(asyncio.to_thread(sync_helper), timeout=1)
        print("after to_thread")
    except BaseException:
        print("to_thread raised/timed out:")
        traceback.print_exc()
    finally:
        logging.getLogger().removeHandler(handler)
        task.cancel()


asyncio.run(main())

I ran that on macOS with Python 3.13:

/opt/homebrew/bin/python3.13 -u repro.py

The failure I saw was:

debug True
consume start
put before from asyncio_0
to_thread raised/timed out:
Traceback (most recent call last):
  ...
  File "asyncio/queues.py", line 170, in put_nowait
    self._wakeup_next(self._getters)
  File "asyncio/queues.py", line 76, in _wakeup_next
    waiter.set_result(None)
  File "asyncio/base_events.py", line 835, in call_soon
    self._check_thread()
  File "asyncio/base_events.py", line 872, in _check_thread
    raise RuntimeError()
RuntimeError: Non-thread-safe operation invoked on an event loop other than the current one

The SDK-level trigger is the same shape:

asyncio.to_thread(sync_helper)
  -> logging.info(...)
  -> LogForwardingHandler.emit(...)
  -> AsyncLogSender.publish(...)
  -> asyncio.Queue.put_nowait(...)

The important details for reproducing are:

• loop.set_debug(True) is enabled.
• AsyncLogSender.consume() / await q.get() is already waiting.
• ctx_step_run_id is set before calling asyncio.to_thread(...).
• The sync function inside to_thread calls logging.info(...).
  If consume() has not reached the pending q.get() yet, or asyncio debug mode is not enabled, this can be harder to observe directly. But the underlying issue is still that asyncio.Queue.put_nowait() is being called from a non-owner thread while the queue is tied to an event loop.

[hobbyhack]

@mrkaye97 Original issue was Linux. I didn't try Windows or MacOS