fix(executor): task refs + pipeline semaphore + off-loop broker [修复组③] - #27
Conversation
…p broker dispatch [C5,C6,C9] - C5: LocalExecutor.dispatch_collection/dispatch_scheduled_collection now hold a strong reference to their asyncio.Task (in new _collection_tasks / _scheduled_collection_tasks dicts, popped on completion) instead of discarding the return of create_task, which asyncio may otherwise GC mid-flight — same fix already in place for dispatch_acquisition. dispatch_scheduled_collection also gains dispatch_acquisition's skip-if-inflight check, since schedule_id (unlike task_id) is not unique per tick. - C6: local.py gains a process-wide asyncio.Semaphore bounding concurrently RUNNING pipeline executions, independent of the per-domain cap in pipeline/domain_limiter.py. The limit is configurable via the new Settings.local_max_concurrent_pipelines field (env LOCAL_MAX_CONCURRENT_PIPELINES, default 8). The semaphore wraps pipeline execution (inside _run_collection/_run_scheduled), not dispatch, so dispatch_collection/dispatch_scheduled_collection stay fast and non-blocking; a task queued behind the semaphore is already referenced per the C5 fix, so it can't be GC'd while waiting. - C9: CeleryExecutor.dispatch_collection/dispatch_scheduled_collection wrap their apply_async calls in asyncio.to_thread so the synchronous broker round-trip no longer runs inline on the event loop. Tests extended in tests/unit/test_executor.py for all three findings.
|
✅ Health: 9.1 📋 At a glance Files & modules (2)
🚨 Change risk: 9.0/10 (high)
🔎 More signals (1)🔥 Hotspots touched (2)
👀 Suggested reviewers @xujinghua 📊 Full report · ⭐ Star Repowise · 📥 Install bot · Last updated 2026-07-18 17:06 UTC |
📝 WalkthroughSummary by CodeRabbit
WalkthroughThe local executor now limits concurrent pipeline runs, retains inflight tasks, and skips overlapping scheduled runs. Celery broker dispatch is offloaded to worker threads, and tests cover configuration, concurrency, task tracking, scheduling, and event-loop responsiveness. ChangesExecutor dispatch behavior
Estimated code review effort: 4 (Complex) | ~45 minutes Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a process-wide concurrency cap for local pipeline executions, ensures strong task references are retained to prevent premature garbage collection, and offloads Celery's synchronous apply_async calls to a separate thread using asyncio.to_thread. The review feedback suggests using weakref.WeakKeyDictionary instead of id(loop) to prevent potential bugs with recycled event loop IDs, which also renders the manual test cleanup fixture obsolete.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| import asyncio | ||
| import logging | ||
|
|
||
| from backend.config import get_settings | ||
| from backend.executor.base import AbstractExecutor | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| # AUDIT C6: process-wide cap on concurrently-RUNNING pipeline executions | ||
| # (manual/webhook dispatch_collection + every dispatch_scheduled_collection), | ||
| # independent of the per-domain cap in pipeline/domain_limiter.py. Keyed by | ||
| # event loop id, mirroring domain_limiter._semaphore, so a semaphore is never | ||
| # reused across event loops (production runs one loop and shares correctly; | ||
| # tests get a fresh loop each and never touch a stale entry). | ||
| _pipeline_semaphores: dict[int, asyncio.Semaphore] = {} | ||
|
|
||
|
|
||
| def _pipeline_semaphore() -> asyncio.Semaphore: | ||
| loop = asyncio.get_running_loop() | ||
| key = id(loop) | ||
| sem = _pipeline_semaphores.get(key) | ||
| if sem is None: | ||
| sem = asyncio.Semaphore(get_settings().local_max_concurrent_pipelines) | ||
| _pipeline_semaphores[key] = sem | ||
| return sem |
There was a problem hiding this comment.
Using id(loop) as a key in a global dictionary is prone to bugs because Python's id() is only guaranteed to be unique during the lifetime of the object. When an event loop is closed and garbage collected (especially during tests or loop restarts), its memory address (and thus its id) can be recycled for a newly created event loop. This causes the new loop to reuse a stale semaphore bound to the old loop, leading to RuntimeError or unexpected concurrency limits.
Using weakref.WeakKeyDictionary with the event loop object itself as the key resolves this cleanly. It automatically removes the semaphore entry when the event loop is garbage collected, preventing both memory leaks and ID recycling bugs without requiring manual cleanup fixtures in tests.
import asyncio
import logging
import weakref
from backend.config import get_settings
from backend.executor.base import AbstractExecutor
logger = logging.getLogger(__name__)
# AUDIT C6: process-wide cap on concurrently-RUNNING pipeline executions
# (manual/webhook dispatch_collection + every dispatch_scheduled_collection),
# independent of the per-domain cap in pipeline/domain_limiter.py. Keyed by
# event loop object using a WeakKeyDictionary, so semaphores are automatically
# cleaned up when loops are closed and garbage collected, preventing memory
# leaks and ID reuse bugs.
_pipeline_semaphores: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Semaphore] = weakref.WeakKeyDictionary()
def _pipeline_semaphore() -> asyncio.Semaphore:
loop = asyncio.get_running_loop()
sem = _pipeline_semaphores.get(loop)
if sem is None:
sem = asyncio.Semaphore(get_settings().local_max_concurrent_pipelines)
_pipeline_semaphores[loop] = sem
return sem| from backend.executor import local as local_executor_module | ||
|
|
||
|
|
||
| @pytest.fixture(autouse=True) | ||
| def _clear_pipeline_semaphore_registry(): | ||
| """AUDIT C6's semaphore registry is a module-level dict keyed by event | ||
| loop id (mirrors backend.pipeline.domain_limiter._semaphores). Clear it | ||
| around every test in this file so a stale entry from a prior test's | ||
| (closed) event loop can never be reused just because CPython recycled | ||
| its id() for a new loop object.""" | ||
| local_executor_module._pipeline_semaphores.clear() | ||
| yield | ||
| local_executor_module._pipeline_semaphores.clear() |
There was a problem hiding this comment.
With the transition of _pipeline_semaphores to a weakref.WeakKeyDictionary, the manual clearing of the registry between tests is no longer necessary. Closed event loops will be garbage collected, and their corresponding semaphore entries will be automatically discarded. We can safely remove this fixture and its import.
| from backend.executor import local as local_executor_module | |
| @pytest.fixture(autouse=True) | |
| def _clear_pipeline_semaphore_registry(): | |
| """AUDIT C6's semaphore registry is a module-level dict keyed by event | |
| loop id (mirrors backend.pipeline.domain_limiter._semaphores). Clear it | |
| around every test in this file so a stale entry from a prior test's | |
| (closed) event loop can never be reused just because CPython recycled | |
| its id() for a new loop object.""" | |
| local_executor_module._pipeline_semaphores.clear() | |
| yield | |
| local_executor_module._pipeline_semaphores.clear() | |
| # The _clear_pipeline_semaphore_registry fixture is no longer needed | |
| # because _pipeline_semaphores now uses a WeakKeyDictionary. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@backend/config.py`:
- Around line 27-34: The local_max_concurrent_pipelines configuration currently
accepts invalid non-positive values. Validate this setting as strictly greater
than zero during configuration loading or validation, preserving the existing
default of 8, and add boundary tests covering zero and negative values being
rejected.
In `@backend/executor/celery_exec.py`:
- Around line 38-51: Update the Celery enqueue flows in the collection dispatch
method and dispatch_scheduled_collection to shield or otherwise await the
in-flight asyncio.to_thread broker call after cancellation, allowing apply_async
to finish before re-raising CancelledError. Preserve the existing task arguments
and return behavior, and add regression coverage verifying cancellation cannot
leave collection or scheduled work enqueued in the background or duplicated on
retry.
In `@backend/executor/local.py`:
- Around line 17-26: Update _pipeline_semaphore and _pipeline_semaphores to key
entries by the asyncio event loop object rather than id(loop), ensuring each
loop retrieves only its own semaphore. Preserve the existing per-loop creation
and configured concurrency limit while avoiding stale semaphore reuse across
closed and newly created loops.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 57783396-2f47-4397-9332-b9fb673faf3a
📒 Files selected for processing (4)
backend/config.pybackend/executor/celery_exec.pybackend/executor/local.pytests/unit/test_executor.py
| # AUDIT C6: process-wide cap on concurrently-RUNNING pipeline executions in | ||
| # the local (in-process asyncio) executor — independent of the per-domain | ||
| # cap in pipeline/domain_limiter.py. Bounds how many schedules firing on | ||
| # the same tick plus manual/webhook triggers can drive Chrome/opencli | ||
| # subprocesses at once on the one event loop. Only meaningful when | ||
| # task_executor="local" (celery fans out across worker processes | ||
| # instead). Env: LOCAL_MAX_CONCURRENT_PIPELINES. | ||
| local_max_concurrent_pipelines: int = 8 |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== backend/config.py ==\n'
ast-grep outline backend/config.py --view expanded || true
printf '\n== relevant settings/tests search ==\n'
rg -n "local_max_concurrent_pipelines|LOCAL_MAX_CONCURRENT_PIPELINES|Semaphore|concurrent pipelines" backend tests . || trueRepository: 2233admin/opencli-admin
Length of output: 5930
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== backend/executor/local.py ==\n'
cat -n backend/executor/local.py
printf '\n== tests/unit/test_executor.py (relevant section) ==\n'
sed -n '180,260p' tests/unit/test_executor.py
printf '\n== runtime probe: asyncio.Semaphore and pydantic settings coercion ==\n'
python3 - <<'PY'
import asyncio
import os
from pydantic import BaseSettings
class S(BaseSettings):
x: int = 8
for value in ["0", "-1", "3"]:
os.environ["X"] = value
try:
s = S()
print(f"env={value!r} -> parsed={s.x}")
sem = asyncio.Semaphore(s.x)
print(f" semaphore._value={sem._value}")
except Exception as e:
print(f"env={value!r} -> ERROR {type(e).__name__}: {e}")
PYRepository: 2233admin/opencli-admin
Length of output: 10628
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== backend/config.py (top) ==\n'
sed -n '1,80p' backend/config.py | cat -n
printf '\n== runtime probe: actual backend.config.Settings parsing ==\n'
python3 - <<'PY'
import os
from contextlib import contextmanager
`@contextmanager`
def setenv(key, value):
old = os.environ.get(key)
os.environ[key] = value
try:
yield
finally:
if old is None:
os.environ.pop(key, None)
else:
os.environ[key] = old
from backend.config import Settings
for value in ["0", "-1", "3"]:
with setenv("LOCAL_MAX_CONCURRENT_PIPELINES", value):
try:
s = Settings()
print(f"value={value!r} -> parsed={s.local_max_concurrent_pipelines!r}")
except Exception as e:
print(f"value={value!r} -> ERROR {type(e).__name__}: {e}")
PYRepository: 2233admin/opencli-admin
Length of output: 4255
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
import asyncio
import inspect
for value in [0, -1, 1]:
try:
sem = asyncio.Semaphore(value)
print(f"Semaphore({value}) -> ok, initial={sem._value}")
except Exception as e:
print(f"Semaphore({value}) -> {type(e).__name__}: {e}")
print("\n--- asyncio.Semaphore source snippet ---")
src = inspect.getsource(asyncio.Semaphore)
print(src)
PYRepository: 2233admin/opencli-admin
Length of output: 3334
Reject non-positive concurrency limits. 0 leaves the local executor semaphore permanently locked, and negatives raise ValueError when it is created; constrain local_max_concurrent_pipelines to > 0 and add boundary tests.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/config.py` around lines 27 - 34, The local_max_concurrent_pipelines
configuration currently accepts invalid non-positive values. Validate this
setting as strictly greater than zero during configuration loading or
validation, preserving the existing default of 8, and add boundary tests
covering zero and negative values being rejected.
| result = await asyncio.to_thread( | ||
| run_collection.apply_async, | ||
| kwargs={"task_id": task_id, "parameters": parameters}, | ||
| ) | ||
| return {"task_id": task_id, "celery_task_id": result.id} | ||
|
|
||
| async def dispatch_scheduled_collection( | ||
| self, schedule_id: str, source_id: str, parameters: dict | ||
| ) -> None: | ||
| from backend.worker.tasks import run_scheduled_collection | ||
| run_scheduled_collection.apply_async( | ||
| kwargs={"schedule_id": schedule_id, "source_id": source_id, "parameters": parameters} | ||
|
|
||
| await asyncio.to_thread( | ||
| run_scheduled_collection.apply_async, | ||
| kwargs={"schedule_id": schedule_id, "source_id": source_id, "parameters": parameters}, |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== backend/executor/celery_exec.py ==\n'
wc -l backend/executor/celery_exec.py
sed -n '1,220p' backend/executor/celery_exec.py
printf '\n== search for celery_exec tests / cancellation ==\n'
rg -n "celery_exec|apply_async|to_thread|cancellation|CancelledError|dispatch_scheduled_collection|dispatch_collection" backend tests . -g '!**/node_modules/**' || true
printf '\n== asyncio.to_thread cancellation probe ==\n'
python3 - <<'PY'
import asyncio, threading, time
events = []
def work():
events.append(("work_start", threading.current_thread().name, time.time()))
time.sleep(0.2)
events.append(("work_end", threading.current_thread().name, time.time()))
return 123
async def main():
task = asyncio.create_task(asyncio.to_thread(work))
await asyncio.sleep(0.05)
task.cancel()
try:
await task
except asyncio.CancelledError:
events.append(("cancelled", time.time()))
await asyncio.sleep(0.3)
print(events)
asyncio.run(main())
PYRepository: 2233admin/opencli-admin
Length of output: 31706
Guard the Celery enqueue against cancellation races. asyncio.to_thread() can keep running after CancelledError, so a retry may enqueue duplicate collection or scheduled work in the background. Let the broker call finish before re-raising cancellation, and add a cancellation regression test.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/executor/celery_exec.py` around lines 38 - 51, Update the Celery
enqueue flows in the collection dispatch method and
dispatch_scheduled_collection to shield or otherwise await the in-flight
asyncio.to_thread broker call after cancellation, allowing apply_async to finish
before re-raising CancelledError. Preserve the existing task arguments and
return behavior, and add regression coverage verifying cancellation cannot leave
collection or scheduled work enqueued in the background or duplicated on retry.
| _pipeline_semaphores: dict[int, asyncio.Semaphore] = {} | ||
|
|
||
|
|
||
| def _pipeline_semaphore() -> asyncio.Semaphore: | ||
| loop = asyncio.get_running_loop() | ||
| key = id(loop) | ||
| sem = _pipeline_semaphores.get(key) | ||
| if sem is None: | ||
| sem = asyncio.Semaphore(get_settings().local_max_concurrent_pipelines) | ||
| _pipeline_semaphores[key] = sem |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate the executor implementation and its tests
git ls-files backend/executor backend | sed -n '1,200p'
# Map the local executor file structure
if [ -f backend/executor/local.py ]; then
echo "=== backend/executor/local.py ==="
cat -n backend/executor/local.py | sed -n '1,220p'
fi
# Find references to the semaphore registry and any test fixture cleanup
rg -n "_pipeline_semaphores|_pipeline_semaphore|get_running_loop|local_max_concurrent_pipelines|clear\\(" backend tests -S
# Find tests touching loop reuse / event loop fixtures
rg -n "event_loop|asyncio\\.Semaphore|loop id|id\\(loop\\)|closed loop|recycled" tests backend -SRepository: 2233admin/opencli-admin
Length of output: 29677
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== backend/pipeline/domain_limiter.py ==="
cat -n backend/pipeline/domain_limiter.py | sed -n '1,140p'
echo "=== tests/unit/test_executor.py (fixture + semaphore tests) ==="
cat -n tests/unit/test_executor.py | sed -n '1,280p'
echo "=== tests/conftest.py (event loop fixture) ==="
cat -n tests/conftest.py | sed -n '1,140p'Repository: 2233admin/opencli-admin
Length of output: 19453
🏁 Script executed:
python3 - <<'PY'
import asyncio, gc, sys
print("python", sys.version)
# Probe 1: loop id recycling after creating/closing many loops.
ids = []
for _ in range(2000):
loop = asyncio.new_event_loop()
ids.append(id(loop))
loop.close()
del loop
gc.collect()
dups = len(ids) - len(set(ids))
print("loop_id_duplicates", dups)
if dups:
seen = set()
for i, x in enumerate(ids):
if x in seen:
print("first_reused_id", x, "at_index", i)
break
seen.add(x)
# Probe 2: create semaphore in one loop, then use it in another loop.
async def make_sem():
return asyncio.Semaphore(0)
loop1 = asyncio.new_event_loop()
try:
sem = loop1.run_until_complete(make_sem())
finally:
loop1.close()
async def try_acquire(s):
await asyncio.wait_for(s.acquire(), timeout=0.1)
return "acquired"
loop2 = asyncio.new_event_loop()
try:
result = loop2.run_until_complete(try_acquire(sem))
print("cross_loop_acquire", result)
except Exception as e:
print("cross_loop_acquire_error", type(e).__name__, str(e))
finally:
loop2.close()
PYRepository: 2233admin/opencli-admin
Length of output: 324
Key the semaphore registry by the loop object, not id(loop). id(loop) is reused quickly, and this registry never evicts entries, so a new loop can pick up a stale semaphore from a closed one. Use the loop object (or a weak-key registry) to keep semaphore ownership tied to loop lifetime.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/executor/local.py` around lines 17 - 26, Update _pipeline_semaphore
and _pipeline_semaphores to key entries by the asyncio event loop object rather
than id(loop), ensuring each loop retrieves only its own semaphore. Preserve the
existing per-loop creation and configured concurrency limit while avoiding stale
semaphore reuse across closed and newly created loops.
1 participant
修复组③ — 执行器 (账本 C5/C6/C9)
Sonnet 实施, Fable 审计通过 (强引用时序 / semaphore 包执行体 / to_thread 逐项核过)。
create_task可被 GC 吞任务 → 强引用 dict (照同类dispatch_acquisition先例)。关键时序: task 入库先于 acquire semaphore, 故阻塞等待期间也被引用, 不只运行时。manual (task_id 唯一) track-and-pop; scheduled (schedule_id 复用) skip-if-inflight 消自重叠asyncio.Semaphore按 event-loop id keyed (照 domain_limiter), 包_run_collection/_run_scheduled执行体不包 dispatch (dispatch 保持快速非阻塞)。上限LOCAL_MAX_CONCURRENT_PIPELINES默认 8, 走 pydantic-settingsapply_async同步 broker 往返阻塞事件循环 → 两处包asyncio.to_threadDeviation (审计已收)
dispatch_acquisition自己的 apply_async 未动 — flag 为 follow-upTest
tests/unit/test_executor.py21/21 (5× 连跑稳定): C5 引用保留+完成弹出、scheduled 在飞跳过、C6 semaphore 真正限流 N>limit、config 默认+env override、C9 经 to_thread