diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index f3f2001a..81aacbba 100644 --- a/backend/druks/workflows.py +++ b/backend/druks/workflows.py @@ -503,8 +503,9 @@ class Workflow: # What this workflow's runs are about, written ``subject = WorkItem`` on the # subclass. None for a workflow about nothing, which says so by silence. subject = _DeclaredSubject(None) - # When set to a cron string, the workflow also registers a schedule that - # fires its run() on that cadence (no subject — a framework cron). + # When set to a cron string, the workflow also registers a schedule. It fires + # dispatch() on that cadence if the class declares one (the policy starts the + # real subject-backed run), else run() with no subject (a framework cron). every: ClassVar[str | None] = None # True holds one warm VM across the run's agent calls (released at gate parks); # False gives each call a throwaway VM. @@ -561,6 +562,19 @@ def __init_subclass__(cls, **kwargs: Any) -> None: f"parameter(s) {sorted(claimed)} — attribution is platform " "routing, not workflow input; name the parameter differently" ) + if cls.every and getattr(cls, "dispatch", None): + required = [ + name + for name, p in inspect.signature(cls.dispatch).parameters.items() + if p.default is inspect.Parameter.empty + and p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD, p.KEYWORD_ONLY) + ] + if required: + raise WorkflowError( + f"{cls.__name__}.dispatch() requires {required}, but a scheduled " + "dispatch fires with no arguments — a schedule's dispatch must be " + "nullary" + ) _wrap_steps(cls) _register_entry(cls) workflows.register(cls) @@ -881,6 +895,15 @@ async def _run_instance( await instance._reap_run() +async def _dispatch_instance(cls: type[Workflow], _context: dict[str, Any] | None = None) -> Any: + # The cron tick runs no workflow of its own kind — no Run row — it just calls + # dispatch(), which start()s the real subject-backed run. Body level, not a + # step: DBOS only allows the child-start inside start() from a workflow body. + # The session gives dispatch()'s reads a transaction, committed on exit. + async with step_session(): + return await cls.dispatch() + + def _register_entry(cls: type[Workflow]) -> None: # The closure binds cls outside the durable arguments: the DBOS workflow # NAME (the kind) is what says which class this is, so recovery rebinds by @@ -892,4 +915,5 @@ async def _entry(subject: dict[str, Any] | None, input: dict[str, Any]) -> None: cls._entry = staticmethod(_entry) # type: ignore[assignment] if cls.every: - register_schedule(cls, partial(_run_instance, cls)) + fire = _dispatch_instance if getattr(cls, "dispatch", None) else _run_instance + register_schedule(cls, partial(fire, cls)) diff --git a/backend/tests/test_durable_sdk.py b/backend/tests/test_durable_sdk.py index 4f08e487..b91e9e50 100644 --- a/backend/tests/test_durable_sdk.py +++ b/backend/tests/test_durable_sdk.py @@ -127,6 +127,19 @@ class DailySweep(Workflow): async def run(self) -> None: # pragma: no cover - not fired in tests SINK.append("swept") + class ScheduledDispatch(Workflow): + # every + a dispatch() classmethod: the tick fires dispatch(), never the + # subjectless run(). dispatch() start()s for real — the enqueue must work + # from the scheduled workflow's body (a step is forbidden the child-start). + subject = Widget + every = "0 */4 * * *" + + @classmethod + async def dispatch(cls) -> str: + return await cls.start(subject=Widget.get_for_subject_id("313131")) + + async def run(self) -> None: ... + class SubjectFlow(Workflow): # Records the subject the platform threaded in, and returns a BaseModel # so the result rides its workflow.finished event. @@ -185,6 +198,7 @@ async def run_multistep(self) -> None: SubjectlessConfirmFlow, ReviewFlow, AttributedFlow, + ScheduledDispatch, ) @@ -217,7 +231,7 @@ def rt(): session.flush() session.add_all( Widget(id=subject_id) - for subject_id in (7, 4242, 636363, 424242, 515151, 878787, 909090) + for subject_id in (7, 4242, 636363, 424242, 515151, 878787, 909090, 313131) ) session.add( HarnessConnection( @@ -243,6 +257,7 @@ def rt(): subjectless_confirm_flow, review_flow, attributed_flow, + scheduled_dispatch, ) = _build_units() os.environ["DRUKS_DATABASE_URL"] = URL init_dbos() @@ -260,6 +275,7 @@ def rt(): SubjectlessConfirmFlow=subjectless_confirm_flow, ReviewFlow=review_flow, AttributedFlow=attributed_flow, + ScheduledDispatch=scheduled_dispatch, ) finally: shutdown() @@ -278,6 +294,7 @@ def rt(): workflows._items.pop("subjectless_confirm_flow", None) workflows._items.pop("review_flow", None) workflows._items.pop("attributed_flow", None) + workflows._items.pop("scheduled_dispatch", None) if db_url_snap is None: os.environ.pop("DRUKS_DATABASE_URL", None) else: @@ -626,6 +643,53 @@ async def test_every_registers_schedule(rt): assert len(params) == 2 and params[1].name == "context" +async def test_scheduled_tick_fires_dispatch_not_run(rt): + # A workflow that declares dispatch() is subject-backed — its run() can't fire + # subjectless. The tick must reach dispatch(), and dispatch()'s start() must + # enqueue the real run from the scheduled workflow's body. + from datetime import UTC, datetime + + from druks.durable.engine import _scheduled + + _, fn = next(row for row in _scheduled if row[0].kind == "scheduled_dispatch") + await fn(datetime.now(UTC), None) + + def dispatched_run(): + session = get_session(rt.engine) + try: + return session.execute( + select(Run).where(Run.kind == "scheduled_dispatch") + ).scalar_one_or_none() + finally: + session.close() + + deadline = asyncio.get_event_loop().time() + 15 + while asyncio.get_event_loop().time() < deadline: + run = dispatched_run() + if run and run.state == RunState.FINISHED: + break + await asyncio.sleep(0.1) + assert run.state == RunState.FINISHED + assert run.subject_label == "W-313131" # about its subject, not subjectless + + +async def test_scheduled_dispatch_must_be_nullary(rt): + # The tick fires dispatch() with no arguments, so a required parameter is a + # declaration error, caught when the class is defined. + from druks.durable.exceptions import WorkflowError + + with pytest.raises(WorkflowError, match="nullary"): + + class NeedsArg(Workflow): + every = "0 6 * * *" + + async def run(self) -> None: ... # pragma: no cover + + @classmethod + async def dispatch(cls, target: str) -> str: # pragma: no cover + return target + + async def test_apply_schedules_drops_undeclared(rt): # A schedule the sys-db still holds but no Workflow declares (a renamed or # removed cron) must be reconciled away, or it keeps firing a dead name. diff --git a/docs/writing-an-extension.md b/docs/writing-an-extension.md index 5bfa13dd..3e7ae9e2 100644 --- a/docs/writing-an-extension.md +++ b/docs/writing-an-extension.md @@ -217,10 +217,28 @@ class Sweep(Workflow): every = "0 6 * * *" ``` -Every parameter of a scheduled workflow needs a default because the scheduler -supplies no input. Druks evaluates cron expressions in the operator timezone. -The dashboard can retune or disable a declared schedule but cannot invent a new -workflow schedule. +The tick fires the workflow's body with no subject and no input, so every body +parameter needs a default. A workflow whose runs are *about* something (it +declares a `subject`) shouldn't fire that way. Give it a `dispatch()` classmethod +and the schedule fires that instead — it resolves the subject and starts the +real run: + +```python +class Engage(Workflow): + subject = Account + every = "0 */4 * * *" + + @classmethod + async def dispatch(cls) -> str: + return await cls.start(subject=Account.get()) + + async def run(self) -> None: + ... +``` + +A scheduled `dispatch()` fires with no arguments, so it must be nullary. Druks +evaluates cron expressions in the operator timezone. The dashboard can retune or +disable a declared schedule but cannot invent a new workflow schedule. A workflow may declare its own operator settings: