tao.chenandClaude
9fc886a55e
schedule:
- New _execution_loop runs alongside _database_event_loop. It claims
job.node.execute rows, sets a 30-min lease on available_at, then
dispatches each as asyncio.create_task under a Semaphore(N).
Polling loop is back to sub-millisecond turnaround for
schedule.run.requested and job.node.finished. Long notebook
execution no longer blocks DAG advance events.
- _process_pending_events filters by event_type IN
('schedule.run.requested', 'job.node.finished'); the executor
loop owns job.node.execute exclusively.
- _process_outbox_event builds a plain dict envelope before
handler dispatch; the previous ORM-row handoff risked
DetachedInstanceError once the outer session closed.
- _sync_once uses get_job + reschedule_job for existing job ids
instead of add_job(replace_existing=True). Each cron schedule
no longer removed-and-readded every 5s.
- service.py threads settings.schedule_execution_concurrency into
the orchestrator (default 4).
common:
- create_async_engine gets explicit pool_size=10, max_overflow=20,
pool_recycle=1800. No more relying on SQLAlchemy defaults.
- New schedule_execution_concurrency setting.
runtime:
- scan_workspaces: add missing 'import os' (NameError on startup)
and switch to asyncio.gather bounded by Semaphore(4) so N
workspaces start in parallel instead of sequentially.
Co-Authored-By: Claude <noreply@anthropic.com>
Schedule Executor
独立调度执行服务，内置 APScheduler。
- Cron 任务持久化到 MySQL 的
apscheduler_jobs表； - FastAPI Backend 创建运行记录和 Outbox 事件后，通过 HTTP 尝试立即推送；
- HTTP 推送失败时，Executor 继续轮询 MySQL
outbox_events，保证任务不会丢失； - Executor 负责 DAG 节点派发、稳定版本执行、重试、状态推进和结果回写；
- 不依赖 Redis，MySQL 是调度状态与幂等状态的唯一权威。