Running enqueued/pending workflows on worker startup

[pmalic]

Presently, to implement a cron-type worker you need to implement a waiting loop, something similar to the example from the Vercel integration docs (https://docs.dbos.dev/integrations/vercel):

async def main():
    DBOS.launch()
    while True:
        await asyncio.sleep(300)
        if not await DBOS.list_queued_workflows_async():
            break

if __name__ == "__main__":
    asyncio.run(main())

Have you considered adding a function (or an optional parameter to DBOS.launch) that would wait until all currently enqueued/pending workflows "dequeue-able" by the current worker are executed?

Using a wait loop with list_queued_workflows can be bug-prone because you could have a queued workflow in a queue not defined in this worker so this loop would run forever without actually dequeueing any workflows. I encountered this problem myself in Python and solved it by adding the queue_name parameter to list_queued_workflows (you could maybe adjust the Vercel example so other people don't run into the same issue :).

Or maybe you could extend the Python docs to include an example of this worker "pattern". Curiously, the LLM prompt page has an example of something related (await asyncio.Event().wait()), but this isn't mentioned in the "human-intended" docs, i.e., the case when you have only your workflow code and nothing else.

[kraftp]

Good timing--I just opened a PR yesterday to add a new example of the worker pattern to the docs: dbos-inc/dbos-docs#476

Good call on adjusting the examples using the list_workflows loop, I'll double-check them and add filters where needed.

[pmalic]

Read the Queue Worker example just now, great! Maybe you can mention this pattern of not having a long-running worker, but starting the worker in some intervals or being triggered by something. Basically the pattern that is described in the Vercel integration docs, because it's far from being Vercel-specific, it could be an actual cron job, or e.g., a Kubernetes or a Google Cloud Run job, created by something that also enqueued the workflow, etc.

While I have your attention may I provide some more feedback about worker-related documentation and background workflow execution in general :)

I read the whole general and Python-specific documentation and I wasn't really sure how to set up a distributed system consisting of several different "types" of workers running a data pipeline (e.g., workers that need to run on separate infrastructure because some steps in the pipeline need a GPU + different steps in the pipeline have different parallelization/concurrency requirements, etc).

After reading the docs, I had a hunch how it should work, but wasn't sure because, although distributed setting and scaling is mentioned in the docs, there isn't a lot of details how to implement it. I had to write a small prototype to verify my "hunch". I was reviewing another workflow orchestration/durable execution system in addition to DBOS which was older and more "established", so had more documentation and examples. I preferred DBOS because it doesn't need a long-running orchestration service so decided to write a prototype. Someone else might rely only on documentation and examples to decide which system to choose/recommend.

These are some of the questions that I couldn't answer with a high degree of certainty while relying only on the currently available documentation:

• What's exactly going on inside the background execution of enqueued workflows. How do I control what workflows the current worker/process will try to execute. I don't think it's "spelled out" in the docs that the background execution will only poll from the queues explicitly defined in code, so you have to define all queues that you want your worker to dequeue from. Also, will the background execution try to execute an undefined workflow from a defined queue (it will :).

• How do I enqueue a workflow intended for a different queue and a different type of worker. I can't define the Queue object and call enqueue on it because it would require the workflow code to be defined as well (thus having the current worker also try to dequeue it). Could you use the DBOS client to enqueue although the docs say it's intended for "external code" (you can :).

• How should you set up the main DBOS config (application name and system database) when you have these different "types" of workers (you could call them "different services", same thing). Should they have the same or different application name, should they share the same system database or not (I opted for different name, but shared system database).

P.S. You can close this issue as far as I'm concerned.

[kraftp]

Read the Queue Worker example just now, great! Maybe you can mention this pattern of not having a long-running worker, but starting the worker in some intervals or being triggered by something. Basically the pattern that is described in the Vercel integration docs, because it's far from being Vercel-specific, it could be an actual cron job, or e.g., a Kubernetes or a Google Cloud Run job, created by something that also enqueued the workflow, etc.

While I have your attention may I provide some more feedback about worker-related documentation and background workflow execution in general :)

I read the whole general and Python-specific documentation and I wasn't really sure how to set up a distributed system consisting of several different "types" of workers running a data pipeline (e.g., workers that need to run on separate infrastructure because some steps in the pipeline need a GPU + different steps in the pipeline have different parallelization/concurrency requirements, etc).

After reading the docs, I had a hunch how it should work, but wasn't sure because, although distributed setting and scaling is mentioned in the docs, there isn't a lot of details how to implement it. I had to write a small prototype to verify my "hunch". I was reviewing another workflow orchestration/durable execution system in addition to DBOS which was older and more "established", so had more documentation and examples. I preferred DBOS because it doesn't need a long-running orchestration service so decided to write a prototype. Someone else might rely only on documentation and examples to decide which system to choose/recommend.

These are some of the questions that I couldn't answer with a high degree of certainty while relying only on the currently available documentation:

* What's exactly going on inside the background execution of enqueued workflows. How do I control what workflows the current worker/process will try to execute. I don't think it's "spelled out" in the docs that the background execution will only poll from the queues explicitly defined in code, so you have to define all queues that you want your worker to dequeue from. Also, will the background execution try to execute an undefined workflow from a defined queue (it will :).

* How do I enqueue a workflow intended for a different queue and a different type of worker. I can't define the `Queue` object and call `enqueue` on it because it would require the workflow code to be defined as well (thus having the current worker also try to dequeue it). Could you use the DBOS client to enqueue although the docs say it's intended for "external code" (you can :).

* How should you set up the main DBOS config (application name and system database) when you have these different "types" of workers (you could call them "different services", same thing). Should they have the same or different application name, should they share the same system database or not (I opted for different name, but shared system database).

P.S. You can close this issue as far as I'm concerned.

Great points, I'll improve the docs! The underlying assumption, which we'll spell out more clearly in the architecture page, is that all workers connected to the same system database have the same code base.

1. Yes, each worker will utilize all queues defined in code, and attempt to dequeue all workflows on those queues.

2. Use a client, as you say.

3. For now, I recommend different application name and different system database for isolation.

One thing I've been thinking about is having the option for workers to explicitly listen to specific queues. We could provide a DBOS.listenQueue([queue1, queue2]) method and, if that is used on a DBOS worker, that worker would only dequeue from those queues. This should make it easier to build multiple heterogeneous workers sharing a code base and system database--each worker can enqueue to any queue using queue.enqueue, but only reads from the explicitly registered queues.

[pmalic]

Wow, there's already a PR with listenQueue, thanks!

In the initial version of your comment that I got in GitHub's email notification you mentioned an affinity system. This got me thinking that it could maybe involve somehow tying executor_id's to queues, because to me this property functions more like an executor class/type (i.e., not supposed to be unique). I mistakenly made this property unique and then figured out I broke workflow recovery :) But this listenQueue is more direct, imperative, so looks better to me.