tao.chen/model-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复调度运行失败

Browse Source
This commit is contained in:
Winnie
2026-08-04 14:09:03 +08:00
parent feae98cc24
commit a2deae2f22
7 changed files with 126 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
schedule/src/schedule/orchestrator.py
+83 -24
View File
@@ -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:
schedule/src/schedule/worker.py
+11 -3
View File
@@ -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",