feat: auth

schedule/src/schedule/main.py

@@ -18,12 +18,12 @@ from schedule.storage_client import SchedulerStorageClient
 async def lifespan(app: Any) -> AsyncIterator[None]:
     engine = create_database_engine(settings.database_url)
     session_factory = create_session_factory(engine)
-    backend_http_client = build_storage_http_client()
+    storage_http_client = build_storage_http_client()
     service = SchedulerService(
         session_factory=session_factory,
-        backend_http_client=backend_http_client,
+        storage_http_client=storage_http_client,
         object_store=build_object_store(),
-        storage_client=SchedulerStorageClient(backend_http_client),
+        storage_client=SchedulerStorageClient(storage_http_client),
         database_url=settings.database_url,
     )
     app.state.scheduler_service = service
@@ -32,7 +32,7 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
         yield
     finally:
         await service.close()
-        await backend_http_client.aclose()
+        await storage_http_client.aclose()
         await engine.dispose()
 
 

schedule/src/schedule/orchestrator.py

@@ -47,20 +47,30 @@ class DispatchOrchestrator:
     - ``_database_event_loop`` drains ``schedule.run.requested`` and
       ``job.node.finished`` events under ``dispatch_lock``. These are
       short, in-line DB transactions.
-    - ``_execution_loop`` claims ``job.node.execute`` events, sets a
-      far-future ``available_at`` as a lease, then dispatches each as
+    - ``_execution_loop`` claims ``job.node.execute`` events whose
+      ``available_at <= utcnow()`` and dispatches each as
       ``asyncio.create_task`` so the polling path is never blocked by
       notebook execution. A semaphore caps concurrent notebooks.
 
     Holds a ``dispatch_lock`` to keep two concurrent drain loops from
     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
+    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
+    too early. Tieing the lease to the actual node timeout closes both
+    cases.
     """
 
-    # Lease window for a claimed ``job.node.execute`` event. If the
-    # process dies mid-execution, the row re-eligible after this many
-    # minutes. The handler is idempotent (it short-circuits on terminal
-    # node states), so safe re-execution.
-    EXECUTION_LEASE = timedelta(minutes=30)
+    # Margin added on top of ``timeout_seconds`` when writing the lease
+    # ``available_at``. Gives the worker time to update the row to a
+    # terminal state before the poll re-picks it.
+    LEASE_SLACK = timedelta(seconds=30)
 
     def __init__(
         self,
@@ -140,11 +150,13 @@ class DispatchOrchestrator:
     ) -> int:
         """Claim ``job.node.execute`` rows and dispatch them as tasks.
 
-        The claim step bumps ``available_at`` to a far-future lease so
-        the polling loop does not re-pick the same row while the
-        dispatched task is still running. Status stays ``pending``;
-        the dispatched task flips it to ``published`` or ``failed``
-        when execution completes.
+        Lease is owned by the row itself (the dispatcher sets
+        ``available_at = utcnow() + node.timeout_seconds + LEASE_SLACK``
+        when the event is enqueued), so this method is a pure
+        read-and-dispatch — no DB writes in the claim step. If the
+        process dies before ``_run_node_execute`` finishes, the row
+        re-eligible once ``available_at`` falls back to now; the worker
+        handler is idempotent (short-circuits on terminal node state).
         """
         async with session_scope(self.session_factory) as session:
             statement = (
@@ -158,17 +170,18 @@ class DispatchOrchestrator:
                 .limit(limit)
             )
             events = list((await session.scalars(statement)).all())
-            claimed: list[tuple[dict[str, Any], str]] = []
-            lease_until = utcnow() + self.EXECUTION_LEASE
-            for item in events:
-                item.available_at = lease_until
-                envelope = {
-                    "event_type": item.event_type,
-                    "event_id": item.event_id,
-                    "trace_id": item.trace_id,
-                    "payload": item.payload_json,
-                }
-                claimed.append((envelope, f"mysql:{item.event_id}"))
+            claimed: list[tuple[dict[str, Any], str]] = [
+                (
+                    {
+                        "event_type": item.event_type,
+                        "event_id": item.event_id,
+                        "trace_id": item.trace_id,
+                        "payload": item.payload_json,
+                    },
+                    f"mysql:{item.event_id}",
+                )
+                for item in events
+            ]
         for envelope, message_id in claimed:
             task = asyncio.create_task(
                 self._run_node_execute(envelope, message_id),
@@ -379,6 +392,20 @@ 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.
+        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",
@@ -387,11 +414,7 @@ class DispatchOrchestrator:
             aggregate_type="schedule_node_run",
             aggregate_id=node_run.node_run_id,
             idempotency_key=f"{node_run.node_run_id}:{attempt_no}",
-            available_at=(
-                utcnow() + timedelta(seconds=delay_seconds)
-                if delay_seconds
-                else None
-            ),
+            available_at=available_at,
             payload={
                 "workspace_id": run.workspace_id,
                 "run_id": run.run_id,

schedule/src/schedule/service.py

@@ -6,13 +6,16 @@ Composes three single-purpose components into one bootable service:
   - :class:`schedule.orchestrator.DispatchOrchestrator` — Outbox polling + DAG
   - :class:`schedule.worker.NodeExecutor` — node-level execution
 
-This module also exposes the two factory functions (``build_object_store`` /
-``build_storage_http_client``) consumed by ``schedule.main`` to construct
-the backing resources that flow into the facade.
+This module also exposes the factory function ``build_object_store``
+consumed by ``schedule.main`` to construct the RustFS S3 client.
 
-The ``SchedulerService`` itself stays small: it wires the three components
-together and implements ``trigger_schedule``, the cron post-back to
-Backend that ``CronScheduler`` calls at every cron tick.
+The :class:`SchedulerService` itself stays small: it wires the three
+components together and implements :meth:`SchedulerService.trigger_schedule`,
+the cron tick handler. The trigger writes the new ``ScheduleRuns`` row
+and ``schedule.run.requested`` outbox event in a single transaction
+via :func:`common.scheduler.create_scheduled_run` — no HTTP call to
+the backend, so no service-to-service auth is needed (the schedule
+service shares the same MySQL via the Docker network).
 """
 
 from __future__ import annotations
@@ -25,6 +28,11 @@ import httpx
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from common.config import settings
+from common.scheduler import (
+    SYSTEM_CRON_USER_ID,
+    TriggerError,
+    create_scheduled_run,
+)
 from common.ids import new_ulid
 
 from schedule.orchestrator import DispatchOrchestrator
@@ -49,13 +57,13 @@ class SchedulerService:
         self,
         *,
         session_factory: async_sessionmaker[AsyncSession],
-        backend_http_client: httpx.AsyncClient,
+        storage_http_client: httpx.AsyncClient,
         object_store: Any,
         storage_client: Any,
         database_url: str,
     ) -> None:
         self.session_factory = session_factory
-        self.backend_http_client = backend_http_client
+        self.storage_http_client = storage_http_client
         self.object_store = object_store
         self.storage_client = storage_client
         self.database_url = database_url
@@ -99,11 +107,17 @@ class SchedulerService:
         await self.cron.close()
 
     async def trigger_schedule(self, schedule_id: str) -> None:
-        """Cron tick callback: post back to Backend to register a new run.
+        """Cron tick callback: write a new ``ScheduleRuns`` row + outbox event.
 
-        Backend writes the ``schedule.run.requested`` Outbox row in the same
-        transaction as the ``ScheduleRuns`` insert; the orchestrator's polling
-        loop will pick it up and start advancing the DAG.
+        Runs in its own session. The cron ``triggered_by`` is the
+        stable :data:`common.scheduler.SYSTEM_CRON_USER_ID` (the
+        bootstrap migration seeds the matching ``Users`` row) — we no
+        longer impersonate the schedule's human creator as the previous
+        header-based implementation did.
+
+        Idempotency key is the cron minute bucket, so re-entering the
+        same tick (e.g. after a brief outage) reuses the existing run
+        via the unique constraint on ``schedule_runs.idempotency_key``.
         """
         async with self.session_factory() as session:
             from sqlalchemy import select
@@ -118,26 +132,29 @@ class SchedulerService:
                 or item.trigger_type != "cron"
             ):
                 return
-            user_id = item.created_by
             workspace_id = item.workspace_id
+
         now = datetime.now(UTC)
         idempotency_key = (
             f"cron:{schedule_id}:{now.strftime('%Y%m%d%H%M')}"
         )
-        response = await self.backend_http_client.post(
-            f"/api/v1/schedules/{schedule_id}/run",
-            headers={
-                "X-User-ID": user_id,
-                "X-Workspace-ID": workspace_id,
-                "X-Request-ID": new_ulid(),
-                "Idempotency-Key": idempotency_key,
-            },
-            json={"reason": "cron"},
-        )
-        if response.is_error:
-            raise RuntimeError(
-                f"backend rejected cron run: {response.status_code} "
-                f"{response.text[:500]}"
+        try:
+            async with self.session_factory() as session:
+                await create_scheduled_run(
+                    session,
+                    schedule_id=schedule_id,
+                    workspace_id=workspace_id,
+                    triggered_by_user_id=SYSTEM_CRON_USER_ID,
+                    trigger_type="cron",
+                    idempotency_key=idempotency_key,
+                    trace_id=new_ulid(),
+                )
+                await session.commit()
+        except TriggerError as exc:
+            # Most likely: idempotency_key collision from a previous
+            # tick in the same minute — silently no-op.
+            LOGGER.info(
+                "cron trigger no-op for schedule %s: %s", schedule_id, exc,
             )
 
     async def process_pending_events(
@@ -175,7 +192,14 @@ def build_object_store() -> Any:
 
 
 def build_storage_http_client() -> httpx.AsyncClient:
-    """Construct the httpx client that talks to Backend's HTTP API."""
+    """Construct the httpx client that talks to Backend's storage API.
+
+    The schedule service no longer needs to call any user-facing
+    endpoint (cron trigger writes directly to the DB now), but the
+    storage endpoints at ``/internal/v1/...`` still live on the
+    backend process and are reached via this client. Auth is not
+    required — the client is bound to the shared Docker network.
+    """
     return httpx.AsyncClient(
         base_url=settings.backend_api_url,
         timeout=httpx.Timeout(60.0),