Support multiple cron expressions

Author: r4victor

src/dstack/_internal/core/models/profiles.py

@@ -170,20 +170,34 @@ def validate_time_window(cls, v: Union[int, str]) -> int:
 
 class Schedule(CoreModel):
     cron: Annotated[
-        Optional[str],
+        Union[List[str], str],
         Field(
             description=(
-                "The UTC time when the run needs to be started specified using POSIX cron syntax"
+                "A cron expression or a list of cron expressions specifying the UTC time when the run needs to be started"
             )
         ),
-    ] = None
+    ]
 
     @validator("cron")
-    def _validate_cron(cls, v) -> Optional[str]:
-        if v is None:
-            return None
-        validate_cron(v)
-        return v
+    def _validate_cron(cls, v: Union[List[str], str]) -> List[str]:
+        if isinstance(v, str):
+            values = [v]
+        else:
+            values = v
+        if len(values) == 0:
+            raise ValueError("At least one cron expression must be specified")
+        for value in values:
+            validate_cron(value)
+        return values
+
+    @property
+    def crons(self) -> List[str]:
+        """
+        Access `cron` attribute as a list.
+        """
+        if isinstance(self.cron, str):
+            return [self.cron]
+        return self.cron
 
 
 class ProfileParams(CoreModel):

src/dstack/_internal/server/services/runs.py

@@ -1253,11 +1253,14 @@ def _remove_job_spec_sensitive_info(spec: JobSpec):
 def _get_next_triggered_at(run_spec: RunSpec) -> Optional[datetime]:
     if run_spec.merged_profile.schedule is None:
         return None
-    cron_trigger = CronTrigger.from_crontab(
-        run_spec.merged_profile.schedule.cron,
-        timezone=timezone.utc,
-    )
-    return cron_trigger.get_next_fire_time(
-        previous_fire_time=None,
-        now=common_utils.get_current_datetime(),
-    )
+    now = common_utils.get_current_datetime()
+    fire_times = []
+    for cron in run_spec.merged_profile.schedule.crons:
+        cron_trigger = CronTrigger.from_crontab(cron, timezone=timezone.utc)
+        fire_times.append(
+            cron_trigger.get_next_fire_time(
+                previous_fire_time=None,
+                now=now,
+            )
+        )
+    return min(fire_times)

src/tests/_internal/server/routers/test_runs.py

@@ -7,6 +7,7 @@
 
 import pytest
 from fastapi.testclient import TestClient
+from freezegun import freeze_time
 from httpx import AsyncClient
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -28,6 +29,7 @@
     InstanceType,
     Resources,
 )
+from dstack._internal.core.models.profiles import Schedule
 from dstack._internal.core.models.resources import Range
 from dstack._internal.core.models.runs import (
     ApplyRunPlanInput,
@@ -1370,6 +1372,41 @@ async def test_updates_run(self, test_db, session: AsyncSession, client: AsyncCl
         assert run.run_spec.configuration.replicas == Range(min=1, max=1)
         assert updated_run.run_spec.configuration.replicas == Range(min=2, max=2)
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)
+    async def test_creates_pending_run_if_run_is_scheduled(
+        self, test_db, session: AsyncSession, client: AsyncClient
+    ):
+        user = await create_user(session=session, global_role=GlobalRole.USER)
+        project = await create_project(session=session, owner=user)
+        await add_project_member(
+            session=session, project=project, user=user, project_role=ProjectRole.USER
+        )
+        repo = await create_repo(session=session, project_id=project.id)
+        run_spec = get_run_spec(
+            run_name="test-run",
+            repo_id=repo.name,
+        )
+        run_spec.configuration.schedule = Schedule(cron=["5 * * * *", "10 * * * *"])
+        with freeze_time(datetime(2023, 1, 2, 3, 9, tzinfo=timezone.utc)):
+            response = await client.post(
+                f"/api/project/{project.name}/runs/apply",
+                headers=get_auth_headers(user.token),
+                json={
+                    "plan": {
+                        "run_spec": json.loads(run_spec.json()),
+                        "current_resource": None,
+                    },
+                    "force": False,
+                },
+            )
+        assert response.status_code == 200, response.json()
+        res = await session.execute(select(RunModel))
+        run = res.scalar()
+        assert run is not None
+        assert run.status == RunStatus.PENDING
+        assert run.next_triggered_at == datetime(2023, 1, 2, 3, 10)
+
 
 class TestSubmitRun:
     @pytest.mark.asyncio