refactor(schedule): move orchestrator to application/ + review fixes (stage 8)

Follow-up to the layered refactor (review-driven):

- Move scheduling/orchestrator.py -> application/orchestrator.py
  (orchestrator is application-level coordination, not a cron-trigger
  primitive; matches the intended target tree)
- Migrate orchestrator re-exports from scheduling/__init__.py to
  application/__init__.py; scheduling/ now exposes only CronScheduler
- Rewrite imports + 5 mock.patch string targets in test_janitor.py and
  the orchestrator import in test_layering.py
- Update docstring refs in application/service.py + execution/worker.py
- Add 4 runner smoke tests (test_layering.py): _limited_log under-limit /
  empty-sentinel / above-MAX_LOG_BYTES truncation; execute_artifact
  rejects unsupported script_type with ValueError
- infrastructure/__init__.py re-exports SchedulerStorageClient so
  ``from schedule.infrastructure import SchedulerStorageClient`` is a
  stable top-level surface
- CLAUDE.md engineering note: extend the commit trail to 7476c27 and
  note the stage-8 orchestrator placement

Zero behavior change; schedule/pyproject.toml untouched. 29 tests green.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
tao.chen
2026-08-21 15:26:27 +08:00
co-authored by Claude
parent 7476c27d81
commit 80130c6d77
9 changed files with 83 additions and 30 deletions
+1 -1
View File
@@ -78,7 +78,7 @@ Hard-won lessons. Read the relevant bullet before touching the named area.
### Schedule service layering (domain / scheduling / application / execution / infrastructure) ### Schedule service layering (domain / scheduling / application / execution / infrastructure)
Lessons from the zero-behavior-change refactor that split the flat 11-file `schedule/src/schedule/` into five subpackages (commits 3118694 → c89132a). Public class names (`SchedulerService`, `CronScheduler`, `DispatchOrchestrator`, `NodeExecutor`, `SchedulerStorageClient`) and `schedule/pyproject.toml` are unchanged. Lessons from the zero-behavior-change refactor that split the flat 11-file `schedule/src/schedule/` into five subpackages (commits 3118694 → 7476c27, with a follow-up `git mv` in stage 8 placing the orchestrator under `application/`). Public class names (`SchedulerService`, `CronScheduler`, `DispatchOrchestrator`, `NodeExecutor`, `SchedulerStorageClient`) and `schedule/pyproject.toml` are unchanged.
- **`python -m schedule.notebook_runner` is a stable external contract.** The worker launches the notebook subprocess with `sys.executable, "-m", "schedule.notebook_runner"`. That `-m` string must never change — so `schedule/notebook_runner.py` survives as a 6-line shim re-exporting `main` from `schedule.execution.runners.notebook`. Don't "clean up" the shim. - **`python -m schedule.notebook_runner` is a stable external contract.** The worker launches the notebook subprocess with `sys.executable, "-m", "schedule.notebook_runner"`. That `-m` string must never change — so `schedule/notebook_runner.py` survives as a 6-line shim re-exporting `main` from `schedule.execution.runners.notebook`. Don't "clean up" the shim.
- **Mock `patch()` string targets and in-function lazy imports are invisible to import-line greps.** `test_janitor.py` patched `"schedule.orchestrator.session_scope"` and `test_worker.py` patched `"schedule.service.build_object_store"`; `worker.py` also had a `from schedule.service import ...` *inside a function body*. When files move, these silently no-op against the old path — and hard-crash once the old file is deleted. After any move, grep the whole tree for the old module path (including tests) and rewrite every hit, not just top-level imports. - **Mock `patch()` string targets and in-function lazy imports are invisible to import-line greps.** `test_janitor.py` patched `"schedule.orchestrator.session_scope"` and `test_worker.py` patched `"schedule.service.build_object_store"`; `worker.py` also had a `from schedule.service import ...` *inside a function body*. When files move, these silently no-op against the old path — and hard-crash once the old file is deleted. After any move, grep the whole tree for the old module path (including tests) and rewrite every hit, not just top-level imports.
+17 -2
View File
@@ -1,9 +1,20 @@
"""Application layer — service assembly for the scheduler. """Application layer — service assembly for the scheduler.
Home of the old flat ``schedule/service.py``: ``SchedulerService`` plus the Houses:
``build_object_store`` / ``build_storage_http_client`` factories.
- ``SchedulerService`` plus the ``build_object_store`` /
``build_storage_http_client`` factories (the old flat ``schedule/service.py``).
- ``DispatchOrchestrator`` and the cron / node event constants (the old
``schedule/orchestrator.py`` — the outbox-polling / DAG coordinator is
application-level glue, not a cron-trigger primitive).
""" """
from schedule.application.orchestrator import (
NODE_EXECUTE_EVENT,
NODE_FINISHED_EVENT,
SCHEDULE_RUN_REQUESTED_EVENT,
DispatchOrchestrator,
)
from schedule.application.service import ( from schedule.application.service import (
SchedulerService, SchedulerService,
build_object_store, build_object_store,
@@ -14,4 +25,8 @@ __all__ = [
"SchedulerService", "SchedulerService",
"build_object_store", "build_object_store",
"build_storage_http_client", "build_storage_http_client",
"DispatchOrchestrator",
"SCHEDULE_RUN_REQUESTED_EVENT",
"NODE_EXECUTE_EVENT",
"NODE_FINISHED_EVENT",
] ]
+2 -2
View File
@@ -3,7 +3,7 @@
Composes three single-purpose components into one bootable service: Composes three single-purpose components into one bootable service:
- :class:`schedule.scheduling.scheduler.CronScheduler` — APScheduler + cron sync loop - :class:`schedule.scheduling.scheduler.CronScheduler` — APScheduler + cron sync loop
- :class:`schedule.scheduling.orchestrator.DispatchOrchestrator` — Outbox polling + DAG - :class:`schedule.application.orchestrator.DispatchOrchestrator` — Outbox polling + DAG
- :class:`schedule.execution.worker.NodeExecutor` — node-level execution - :class:`schedule.execution.worker.NodeExecutor` — node-level execution
This module also exposes the factory function ``build_object_store`` This module also exposes the factory function ``build_object_store``
@@ -36,7 +36,7 @@ from common.storage import create_storage
from loguru import logger from loguru import logger
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from schedule.scheduling.orchestrator import DispatchOrchestrator from schedule.application.orchestrator import DispatchOrchestrator
from schedule.scheduling.scheduler import CronScheduler from schedule.scheduling.scheduler import CronScheduler
from schedule.execution.worker import NodeExecutor from schedule.execution.worker import NodeExecutor
+1 -1
View File
@@ -1,6 +1,6 @@
"""Node-level worker: executes one ``job.node.execute`` event. """Node-level worker: executes one ``job.node.execute`` event.
The orchestrator (see ``schedule.scheduling.orchestrator``) writes a ``job.node.execute`` The orchestrator (see ``schedule.application.orchestrator``) writes a ``job.node.execute``
Outbox row with all the metadata needed to run the node (script type, Outbox row with all the metadata needed to run the node (script type,
artifact location, timeout, arguments ...). The polling loop picks those up artifact location, timeout, arguments ...). The polling loop picks those up
and calls :meth:`NodeExecutor.handle_node_execute`. This module owns the and calls :meth:`NodeExecutor.handle_node_execute`. This module owns the
@@ -1 +1,5 @@
"""Infrastructure layer — external I/O adapters (storage API).""" """Infrastructure layer — external I/O adapters (storage API)."""
from schedule.infrastructure.storage import SchedulerStorageClient
__all__ = ["SchedulerStorageClient"]
+6 -17
View File
@@ -1,22 +1,11 @@
"""Scheduling layer — cron trigger + DAG orchestration. """Scheduling layer — cron trigger.
Home of the old flat ``schedule/scheduler.py`` (``CronScheduler``) and Home of the old flat ``schedule/scheduler.py`` (``CronScheduler``).
``schedule/orchestrator.py`` (``DispatchOrchestrator``). Classes moved The class moved byte-identical in the layered refactor; name unchanged.
byte-identical in the layered refactor; names unchanged. The outbox-polling orchestrator used to live here too, but it is an
application-level coordinator — see ``schedule.application.orchestrator``.
""" """
from schedule.scheduling.orchestrator import (
NODE_EXECUTE_EVENT,
NODE_FINISHED_EVENT,
SCHEDULE_RUN_REQUESTED_EVENT,
DispatchOrchestrator,
)
from schedule.scheduling.scheduler import CronScheduler from schedule.scheduling.scheduler import CronScheduler
__all__ = [ __all__ = ["CronScheduler"]
"CronScheduler",
"DispatchOrchestrator",
"SCHEDULE_RUN_REQUESTED_EVENT",
"NODE_EXECUTE_EVENT",
"NODE_FINISHED_EVENT",
]
+6 -6
View File
@@ -23,7 +23,7 @@ import pytest
from common.db.models import ScheduleNodeRuns from common.db.models import ScheduleNodeRuns
from common.eventing import utcnow from common.eventing import utcnow
from schedule.scheduling.orchestrator import NODE_FINISHED_EVENT, DispatchOrchestrator from schedule.application.orchestrator import NODE_FINISHED_EVENT, DispatchOrchestrator
def _make_orchestrator() -> DispatchOrchestrator: def _make_orchestrator() -> DispatchOrchestrator:
@@ -195,7 +195,7 @@ async def test_reap_kills_running_row_past_deadline() -> None:
fake_session.add = MagicMock() fake_session.add = MagicMock()
with patch( with patch(
"schedule.scheduling.orchestrator.session_scope", "schedule.application.orchestrator.session_scope",
return_value=_open_session_scope(fake_session), return_value=_open_session_scope(fake_session),
): ):
killed = await orch._reap_stuck_node_runs() killed = await orch._reap_stuck_node_runs()
@@ -223,7 +223,7 @@ async def test_reap_skips_healthy_row() -> None:
fake_session.add = MagicMock() fake_session.add = MagicMock()
with patch( with patch(
"schedule.scheduling.orchestrator.session_scope", "schedule.application.orchestrator.session_scope",
return_value=_open_session_scope(fake_session), return_value=_open_session_scope(fake_session),
): ):
killed = await orch._reap_stuck_node_runs() killed = await orch._reap_stuck_node_runs()
@@ -266,7 +266,7 @@ async def test_reap_processes_multiple_rows_in_one_pass() -> None:
fake_session.add = MagicMock() fake_session.add = MagicMock()
with patch( with patch(
"schedule.scheduling.orchestrator.session_scope", "schedule.application.orchestrator.session_scope",
return_value=_open_session_scope(fake_session), return_value=_open_session_scope(fake_session),
): ):
killed = await orch._reap_stuck_node_runs() killed = await orch._reap_stuck_node_runs()
@@ -289,7 +289,7 @@ async def test_janitor_loop_propagates_cancellation() -> None:
raise asyncio.CancelledError raise asyncio.CancelledError
with patch( with patch(
"schedule.scheduling.orchestrator.asyncio.sleep", "schedule.application.orchestrator.asyncio.sleep",
side_effect=cancel_on_sleep, side_effect=cancel_on_sleep,
): ):
with pytest.raises(asyncio.CancelledError): with pytest.raises(asyncio.CancelledError):
@@ -321,7 +321,7 @@ async def test_janitor_loop_continues_after_reap_exception() -> None:
raise asyncio.CancelledError raise asyncio.CancelledError
with patch( with patch(
"schedule.scheduling.orchestrator.asyncio.sleep", "schedule.application.orchestrator.asyncio.sleep",
side_effect=_count_sleeps, side_effect=_count_sleeps,
): ):
with pytest.raises(asyncio.CancelledError): with pytest.raises(asyncio.CancelledError):
+46 -1
View File
@@ -36,7 +36,7 @@ from schedule.domain.context import (
from schedule.domain.execution import ExecutionResult from schedule.domain.execution import ExecutionResult
from schedule.execution import NodeExecutor from schedule.execution import NodeExecutor
from schedule.infrastructure.storage import SchedulerStorageClient from schedule.infrastructure.storage import SchedulerStorageClient
from schedule.scheduling.orchestrator import DispatchOrchestrator from schedule.application.orchestrator import DispatchOrchestrator
from schedule.scheduling.scheduler import CronScheduler from schedule.scheduling.scheduler import CronScheduler
@@ -175,3 +175,48 @@ def test_notebook_runner_shim_reexports_real_main() -> None:
from schedule.notebook_runner import main as shim_main from schedule.notebook_runner import main as shim_main
assert shim_main is real.main assert shim_main is real.main
# ── execution/runners: stage-4 merge boundary smoke (F4) ──────────────────
def test_limited_log_under_limit_returns_encoded_unchanged() -> None:
from schedule.execution.runners import notebook
assert notebook._limited_log("hello") == b"hello"
def test_limited_log_empty_returns_sentinel() -> None:
from schedule.execution.runners import notebook
assert notebook._limited_log("") == b"execution produced no console output\n"
def test_limited_log_above_max_bytes_truncates_with_suffix() -> None:
from schedule.execution.runners import notebook
sentinel = b"\n[log truncated by scheduler worker]\n"
overflowing = "x" * (notebook.MAX_LOG_BYTES + 1)
truncated = notebook._limited_log(overflowing)
assert len(truncated) == notebook.MAX_LOG_BYTES
assert truncated.endswith(sentinel)
@pytest.mark.asyncio
async def test_execute_artifact_unsupported_script_type_raises() -> None:
"""``script_type`` outside ``{"notebook", "python"}`` must raise —
otherwise the worker would silently drop a malformed run into the
queue and the node would hang.
"""
from schedule.execution.runners import notebook
with pytest.raises(ValueError, match="unsupported script_type"):
await notebook.execute_artifact(
source=b"print('hi')\n",
run_id="01RUN0000000000000000000A",
node_run_id="01NODE000000000000000000A",
script_type="bogus",
artifact_path="x.py",
arguments=[],
timeout_seconds=30,
)