Prevent overlapping runs of a schedule (dedup across ticks)

[Giuzzilla]

I have some schedules triggering jobs relying on the fact that just one of them must be running at a time per-schedule (/per-workflow), while also being enqueued on a queue limiting overall concurrency across jobs.

Right now, I'm forcing this with a parent workflow triggering the actual workflow with a constant deduplication id:

@DBOS.workflow()
async def my_workflow(scheduled_time: datetime) -> None:
    # do something
    pass

@DBOS.workflow()
async def my_workflow_parent(
    scheduled_time: datetime, context: Any
) -> None:
    with SetEnqueueOptions(deduplication_id="my-workflow"):
        await DBOS.enqueue_workflow_async(
            QUEUE_NAME,
            my_workflow,
            scheduled_time,
        )


MY_WORKFLOW_SCHEDULE: ScheduleInput = {
    "schedule_name": "my-workflow",
    "workflow_fn": my_workflow_parent,
    "schedule": "*/1 * * * *",
    "context": None,
    "cron_timezone": "UTC",
    "queue_name": QUEUE_NAME,
}

An alternative implementation without changes on DBOS-side would be to use two layers of queueing (more complex than the above).

It'd be nice to have one of these things directly on the library:

• The possibility to specify a fixed deduplication_id on a schedule (should be easy)
• Alternatively, a deduplicate flag on the schedule, preventing piling up of jobs (generating an internal deduplication_id).
• I guess this could be even more complex, like choosing between Allow (current behaviour) | Forbid/Skip (maybe with a custom state?) | Replace (cancel+enqueue). See Kubernetes CronJob concurrencyPolicy
• Some other way to serialize schedules, similarly (but not exactly) to depends_on_past in Airflow DAGs

[kraftp]

Got it! To be clear, what semantics exactly are you looking for? The simplest fix for this would be to configure the schedule with a queue with concurrency=1, which guarantees nonoverlapping sequential execution.

[pratikscfr]

Hi @kraftp , can I take this up?

[kraftp]

It's not yet clear to me there's an issue to solve here. We could add complex semantics, but are they needed? Would a concurrency=1 solve this problem by itself?

[Giuzzilla]

Got it! To be clear, what semantics exactly are you looking for? The simplest fix for this would be to configure the schedule with a queue with concurrency=1, which guarantees nonoverlapping sequential execution.

Suppose I have another queue which enforces some other kind of global rate limiting/concurrency across jobs, but I want this particular job to have one run at a time (on top of those limits):

As of now, I can either do two levels of queueing (like the examples in the docs) or do what I did in the example (i.e. an intermediate workflow that just enqueues the actual job with a deduplication id).

The proposals I have made in the issue are not about any bug, just some ideas for a slightly more friendly syntax for applying common patterns in scheduling.

I guess the absolutely easiest implementation would be to allow passing a fixed "deduplication_id" to the schedule itself (which is just an enqueueing option), which would make it work with 1 workflow only.

[pratikscfr]

@kraftp , The concurrency=1 queue suggestion doesn't solve the case where you need per-schedule dedup on top of a global queue's rate limiting. The simplest fix is to expose deduplication_id as an optional field on ScheduleInput — the queue system already has full dedup support, the scheduler just needs to pass it through. This avoids the boilerplate of a parent workflow and is more discoverable than the two-layer queueing workaround.

Your thoughts @kraftp , @Giuzzilla ?

[kraftp]

I'll think about this. The issue with a deduplication_id is that overlapping runs will just disappear, which may be unintuitive. The solution is probably a larger system for controlling what happens to overlapping runs (delay them or don't run them or what have you), but that's a fair amount of complexity. I'll keep thinking.