From a2deae2f22a19d2775b0395f5206b8ae5ddb3cbc Mon Sep 17 00:00:00 2001 From: Winnie <3308978791@qq.com> Date: Tue, 4 Aug 2026 14:09:03 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E8=B0=83=E5=BA=A6=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E5=A4=B1=E8=B4=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 1 + common/src/common/config.py | 7 ++ common/src/common/eventing.py | 21 ++++- common/src/common/scheduler/trigger.py | 4 +- docker-compose.yml | 2 + schedule/src/schedule/orchestrator.py | 107 +++++++++++++++++++------ schedule/src/schedule/worker.py | 14 +++- 7 files changed, 126 insertions(+), 30 deletions(-) diff --git a/.env.example b/.env.example index 2d6ab09..e11629a 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,5 @@ COMPOSE_PROJECT_NAME=model-platform-develop +SCHEDULE_EVENT_NAMESPACE=model-platform-develop # External port of the Nginx gateway. Only Nginx is exposed to the host # (architecture §2.2); backend/runtime/schedule stay on the Docker internal diff --git a/common/src/common/config.py b/common/src/common/config.py index 290c2e7..6de0cb8 100644 --- a/common/src/common/config.py +++ b/common/src/common/config.py @@ -133,6 +133,13 @@ class Settings(BaseSettings): ) # ── schedule execution tuning ──────────────────────────────── + schedule_event_namespace: str = Field( + default="model-platform-local", + description=( + "Namespace prepended to schedule outbox event types. Each " + "deployment that shares a database must use a unique value." + ), + ) schedule_execution_concurrency: int = Field( default=4, description=( diff --git a/common/src/common/eventing.py b/common/src/common/eventing.py index dc8adc0..3003ec1 100644 --- a/common/src/common/eventing.py +++ b/common/src/common/eventing.py @@ -5,6 +5,7 @@ from typing import Any from sqlalchemy.ext.asyncio import AsyncSession +from common.config import settings from common.db.models import OutboxEvents from common.ids import new_ulid @@ -20,6 +21,19 @@ def event_time(value: datetime | None = None) -> str: return item.astimezone(UTC).isoformat().replace("+00:00", "Z") +def schedule_event_type(event_type: str) -> str: + """Return the deployment-scoped schedule event type. + + Multiple development deployments currently share one MySQL database. + Scoping the event type prevents a scheduler from another deployment from + claiming and acknowledging work that only this deployment can execute. + """ + namespace = settings.schedule_event_namespace.strip().strip(".") + if not namespace: + raise ValueError("SCHEDULE_EVENT_NAMESPACE must not be empty") + return f"{namespace}.{event_type}" + + async def add_outbox_event( session: AsyncSession, *, @@ -50,5 +64,10 @@ async def add_outbox_event( return item -__all__ = ["add_outbox_event", "event_time", "utcnow"] +__all__ = [ + "add_outbox_event", + "event_time", + "schedule_event_type", + "utcnow", +] diff --git a/common/src/common/scheduler/trigger.py b/common/src/common/scheduler/trigger.py index 6347010..2b26179 100644 --- a/common/src/common/scheduler/trigger.py +++ b/common/src/common/scheduler/trigger.py @@ -32,7 +32,7 @@ from common.db.models import ( Scripts, Versions, ) -from common.eventing import add_outbox_event, utcnow +from common.eventing import add_outbox_event, schedule_event_type, utcnow from common.ids import new_ulid @@ -337,7 +337,7 @@ async def create_scheduled_run( await add_outbox_event( session, - event_type="schedule.run.requested", + event_type=schedule_event_type("schedule.run.requested"), producer="platform-api", trace_id=trace_id, aggregate_type="schedule_run", diff --git a/docker-compose.yml b/docker-compose.yml index 117cec6..3cecb15 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -54,6 +54,7 @@ services: environment: DATABASE_URL: ${DATABASE_URL:?DATABASE_URL is required} SERVICE_NAME: model-platform-backend + SCHEDULE_EVENT_NAMESPACE: ${SCHEDULE_EVENT_NAMESPACE:-model-platform-local} JWT_SECRET: ${JWT_SECRET:?JWT_SECRET is required} DEMO_AUTH_ENABLED: ${DEMO_AUTH_ENABLED:-false} INITIAL_ADMIN_PASSWORD: ${INITIAL_ADMIN_PASSWORD:-admin12345} @@ -133,6 +134,7 @@ services: environment: DATABASE_URL: ${DATABASE_URL:?DATABASE_URL is required} SERVICE_NAME: schedule-executor + SCHEDULE_EVENT_NAMESPACE: ${SCHEDULE_EVENT_NAMESPACE:-model-platform-local} BACKEND_API_URL: http://backend:8000 RUSTFS_ENDPOINT: ${RUSTFS_ENDPOINT:?RUSTFS_ENDPOINT is required} RUSTFS_ACCESS_KEY: ${RUSTFS_ACCESS_KEY:?RUSTFS_ACCESS_KEY is required} diff --git a/schedule/src/schedule/orchestrator.py b/schedule/src/schedule/orchestrator.py index e7d45b2..bcc8560 100644 --- a/schedule/src/schedule/orchestrator.py +++ b/schedule/src/schedule/orchestrator.py @@ -27,7 +27,7 @@ from common.db.models import ( ScheduleNodeRuns, ScheduleRuns, ) -from common.eventing import add_outbox_event, utcnow +from common.eventing import add_outbox_event, schedule_event_type, utcnow from common.ids import new_ulid from schedule.context import ( @@ -38,6 +38,10 @@ from schedule.context import ( LOGGER = logging.getLogger(__name__) +SCHEDULE_RUN_REQUESTED_EVENT = schedule_event_type("schedule.run.requested") +NODE_EXECUTE_EVENT = schedule_event_type("job.node.execute") +NODE_FINISHED_EVENT = schedule_event_type("job.node.finished") + class DispatchOrchestrator: """Polls Outbox + advances DAG schedule runs. @@ -56,10 +60,11 @@ class DispatchOrchestrator: fighting over the same batch. Lease semantics live on the outbox row itself, not in the claim - step. The dispatcher sets ``available_at = utcnow() + node_timeout - + LEASE_SLACK`` when it writes the ``job.node.execute`` event, so a - process crash mid-execution lets the row re-eligible automatically - once the lease expires. A hard-coded 30-minute lease was the + step. A new ``job.node.execute`` event is immediately eligible; + the claim step atomically moves ``available_at`` to ``utcnow() + + node_timeout + LEASE_SLACK``, so a process crash mid-execution lets + the row become eligible again once the lease expires. A hard-coded + 30-minute lease was the original P0-2 bug: a node with ``timeout_seconds = 86_400`` would be re-claimed at 30 minutes and run twice; a node with ``timeout_seconds = 60`` would have its lease expire 29 minutes @@ -163,13 +168,25 @@ class DispatchOrchestrator: select(OutboxEvents) .where( OutboxEvents.event_status == "pending", - OutboxEvents.event_type == "job.node.execute", + OutboxEvents.event_type == NODE_EXECUTE_EVENT, OutboxEvents.available_at <= utcnow(), ) .order_by(OutboxEvents.created_at) .limit(limit) + .with_for_update(skip_locked=True) ) events = list((await session.scalars(statement)).all()) + now = utcnow() + for item in events: + timeout_seconds = max( + 1, + int(item.payload_json.get("timeout_seconds", 1)), + ) + item.available_at = ( + now + + timedelta(seconds=timeout_seconds) + + self.LEASE_SLACK + ) claimed: list[tuple[dict[str, Any], str]] = [ ( { @@ -246,7 +263,10 @@ class DispatchOrchestrator: OutboxEvents.event_status == "pending", OutboxEvents.available_at <= utcnow(), OutboxEvents.event_type.in_( - ("schedule.run.requested", "job.node.finished"), + ( + SCHEDULE_RUN_REQUESTED_EVENT, + NODE_FINISHED_EVENT, + ), ), ) .order_by(OutboxEvents.created_at) @@ -276,8 +296,8 @@ class DispatchOrchestrator: async def _process_outbox_event(self, item: OutboxEvents) -> None: handlers = { - "schedule.run.requested": self._handle_run_requested, - "job.node.finished": self._handle_node_finished, + SCHEDULE_RUN_REQUESTED_EVENT: self._handle_run_requested, + NODE_FINISHED_EVENT: self._handle_node_finished, } handler = handlers.get(item.event_type) if handler is None: @@ -336,7 +356,7 @@ class DispatchOrchestrator: event: dict[str, Any], message_id: str, ) -> None: - if event.get("event_type") != "schedule.run.requested": + if event.get("event_type") != SCHEDULE_RUN_REQUESTED_EVENT: raise ValueError("unexpected event type") payload = event["payload"] async with session_scope(self.session_factory) as session: @@ -360,6 +380,11 @@ class DispatchOrchestrator: run.run_status = "running" run.started_at = utcnow() run.state_version += 1 + await self._bootstrap_root_nodes( + session, + run, + trace_id=event["trace_id"], + ) await self._advance_run( session, run, @@ -392,29 +417,23 @@ class DispatchOrchestrator: ), ) session.add(node_run) - # Lease is owned by the outbox row, not the claim step. Pick - # the later of (now, scheduled retry) and the timeout + slack, - # so a slow node isn't re-dispatched while it's still running - # but a crashed node does become eligible again after its - # timeout expires. See P0-2 in the auth refactor plan. + # ``available_at`` is the first execution time here. The claim + # step moves it forward by timeout + slack to become the retry + # lease while the worker is running. retry_at = ( utcnow() + timedelta(seconds=delay_seconds) if delay_seconds else utcnow() ) - lease_at = utcnow() + timedelta( - seconds=int(node["timeout_seconds"]) - ) + self.LEASE_SLACK - available_at = max(retry_at, lease_at) await add_outbox_event( session, - event_type="job.node.execute", + event_type=NODE_EXECUTE_EVENT, producer="schedule-orchestrator", trace_id=trace_id, aggregate_type="schedule_node_run", aggregate_id=node_run.node_run_id, idempotency_key=f"{node_run.node_run_id}:{attempt_no}", - available_at=available_at, + available_at=retry_at, payload={ "workspace_id": run.workspace_id, "run_id": run.run_id, @@ -431,6 +450,44 @@ class DispatchOrchestrator: ) return node_run + async def _bootstrap_root_nodes( + self, + session: AsyncSession, + run: ScheduleRuns, + *, + trace_id: str, + ) -> None: + """Persist the first runnable nodes before normal DAG advancement.""" + existing_node_id = await session.scalar( + select(ScheduleNodeRuns.node_run_id) + .where(ScheduleNodeRuns.run_id == run.run_id) + .limit(1) + ) + if existing_node_id is not None: + return + + snapshot = run.schedule_snapshot + nodes = snapshot.get("nodes", []) + target_node_ids = { + edge["target_node_id"] for edge in snapshot.get("edges", []) + } + roots = [ + node for node in nodes + if node["node_id"] not in target_node_ids + ] + max_concurrency = max(1, int(snapshot.get("max_concurrency", 1))) + for node in roots[:max_concurrency]: + await self._dispatch_node( + session, + run=run, + node=node, + attempt_no=1, + trace_id=trace_id, + ) + # The shared session factory disables autoflush. Flush here so + # _advance_run observes the roots and cannot dispatch duplicates. + await session.flush() + async def _advance_run( self, session: AsyncSession, @@ -553,8 +610,10 @@ class DispatchOrchestrator: # parents_blocked branch above handles the case # where every parent is terminal but at least one # failed. - len(parent_runs) > 0 - and all( + # ``all([])`` is intentionally true: a root node has + # no parents and must be eligible for the initial + # dispatch that starts the DAG. + all( item is not None and item.node_status == "succeeded" for item in parent_runs @@ -627,7 +686,7 @@ class DispatchOrchestrator: event: dict[str, Any], message_id: str, ) -> None: - if event.get("event_type") != "job.node.finished": + if event.get("event_type") != NODE_FINISHED_EVENT: raise ValueError("unexpected event type") payload = event["payload"] async with session_scope(self.session_factory) as session: diff --git a/schedule/src/schedule/worker.py b/schedule/src/schedule/worker.py index d658315..bf9d4db 100644 --- a/schedule/src/schedule/worker.py +++ b/schedule/src/schedule/worker.py @@ -33,13 +33,21 @@ from common.db.models import ( Versions, Workspaces, ) -from common.eventing import add_outbox_event, event_time, utcnow +from common.eventing import ( + add_outbox_event, + event_time, + schedule_event_type, + utcnow, +) from schedule.context import TERMINAL_NODE_STATES from schedule.execution import ExecutionResult, execute_artifact LOGGER = logging.getLogger(__name__) +NODE_EXECUTE_EVENT = schedule_event_type("job.node.execute") +NODE_FINISHED_EVENT = schedule_event_type("job.node.finished") + class NodeExecutor: """Owns the actual execution of one schedule node (notebook / python).""" @@ -60,7 +68,7 @@ class NodeExecutor: event: dict[str, Any], message_id: str, ) -> None: - if event.get("event_type") != "job.node.execute": + if event.get("event_type") != NODE_EXECUTE_EVENT: raise ValueError("unexpected event type") payload = event["payload"] if not await self._set_node_running(event, message_id): @@ -159,7 +167,7 @@ class NodeExecutor: node_run.state_version += 1 await add_outbox_event( session, - event_type="job.node.finished", + event_type=NODE_FINISHED_EVENT, producer="job-worker", trace_id=event["trace_id"], aggregate_type="schedule_node_run",