[erezrokah]

Child resolver pools used the raw queue_size argument and every pool bounded its executor work queue, which could deadlock a sync or block dropping a scheduler, so this sizes every pool from the computed queue size and applies that limit as a semaphore before submit while keeping the executor's work queue unbounded.

[disq]

Needs changes. This fixes the depth-2 deadlock and the GC/exit hang at the default queue size, but bounding the depth-0 queue adds a new deadlock, and the underlying cause is still there.

1. New deadlock: more top-level work than the depth-0 queue holds. With the default queue_size=0, the depth-0 queue used to be unbounded; it is now concurrency * 100, which is worth calling out in the description. A sync with more resolver × client submissions than that, on tables with relations, deadlocks mid-sync. With Scheduler(2) and one table multiplexed over 1000 clients with one child table (repro below), main completes the sync (and then hangs at interpreter exit), while this branch deadlocks in 3 of 3 runs. The thread dump shows _sync blocked in queue.put inside ThreadPoolExecutor.submit, and both depth-0 workers blocked entering submit for their child resolvers.

2. The root cause stays at every depth. ThreadPoolExecutor.submit calls self._work_queue.put(w) while holding self._shutdown_lock and the module-wide _global_shutdown_lock. A full bounded queue therefore blocks the lock holder and every other pool's submit, including the workers that would drain it. Bigger queues make this rarer at depth 1 and below but don't prevent it: Scheduler(4, queue_size=1) with parent → child → grandchild tables still deadlocks on this branch.

3. A small explicit queue_size can still hang on GC or exit (from reading CPython, not reproduced). The executor's weakref callback and _python_exit put one None per worker on the work queue, which can block when queue_size is smaller than the worker count. The new test only covers the default size.

Suggestion: keep the executor's unbounded work queue and apply backpressure before submit, outside those locks. Each level only waits on the level below it, so there is no cycle. With this change, the repros for 1 and 2 pass, and so does the full suite:

class ThreadPoolExecutorWithQueueSizeLimit(futures.ThreadPoolExecutor):
    def __init__(self, maxsize, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.maxsize = maxsize
        self._slots = threading.BoundedSemaphore(maxsize + self._max_workers)

    def submit(self, fn, /, *args, **kwargs):
        self._slots.acquire()
        try:
            future = super().submit(fn, *args, **kwargs)
        except BaseException:
            self._slots.release()
            raise
        future.add_done_callback(lambda _: self._slots.release())
        return future

Tests: they catch the original bug, as they hang on main. On a regression, though, the stuck worker threads keep pytest from exiting, so CI hangs until its timeout instead of failing. test_default_queue_size_is_used_for_all_depths reads the private _work_queue.maxsize, which the suggestion above would change.

Repro for 1

import pyarrow as pa
from cloudquery.sdk.scheduler import Scheduler, TableResolver
from cloudquery.sdk.schema import Column
from cloudquery.sdk.schema.table import Table


class C:
    def __init__(self, i):
        self.i = i

    def id(self):
        return f"c{self.i}"


class Child(TableResolver):
    def __init__(self):
        super().__init__(Table("child", [Column("c", pa.int64())]))

    def resolve(self, client, parent):
        yield {"c": 1}


class Top(TableResolver):
    def __init__(self, n):
        super().__init__(Table("top", [Column("id", pa.int64())]), child_resolvers=[Child()])
        self.n = n

    def multiplex(self, client):
        return [C(i) for i in range(self.n)]

    def resolve(self, client, parent):
        for i in range(3):
            yield {"id": i}


s = Scheduler(2)  # depth-0 queue of 200
print("messages", sum(1 for _ in s.sync(C(-1), [Top(1000)])))

[disq]

LGTM. All three issues from my earlier review are fixed. I verified them on fa9a120:

• More top-level work than the depth-0 queue holds no longer deadlocks the sync.
• Relations with a full queue no longer deadlock, at any depth.
• Dropping a scheduler with a small explicit queue_size no longer hangs. That one reproduces on the previous version of this PR.

The scenario tests catch them: against the previous bounded-queue version, the three matching scenarios time out and fail. Because they run in subprocesses with a timeout, a regression now fails CI instead of hanging it. The full suite passes, and the branch merges cleanly with main, which now includes #417.

Two non-blocking nits:

1. A one-line comment on why the limit is a semaphore taken before submit() would stop someone from "simplifying" it back to a bounded queue. Something like: "submit() puts on the work queue while holding a lock every pool shares, so a full bounded queue would block other pools' workers too."
2. self._max_workers is a private ThreadPoolExecutor attribute. It's stable in CPython 3.11 to 3.14, but the pool could keep its own copy of max_workers instead.