Convert the v03 pipeline queue to a real queue #1136

nvnieuwk · 2025-08-05T15:28:14Z

This PR implements actual queuing into the v03 pipeline by expanding the current system.

Main changes

Whenever a valid POST request is being sent to the loading_pipeline_enqueue route, a new queue file is created with a name structured as follows: request_<uuid>.json where <uuid> is a unique identifier to prevent any filename collisions
The pipeline worker will check if the queue directory is empty, when not empty, the worker will take the oldest queue file and run the job corresponding to that file. When the job is done, the file gets deleted and the next one is started

bpblanken · 2025-08-05T16:46:16Z

v03_pipeline/api/app_test.py

@@ -92,6 +92,7 @@ async def test_loading_pipeline_enqueue(self):
                    'projects_to_run': ['project_a'],
                    'reference_genome': 'GRCh38',
                    'sample_type': 'WGS',
+                    'skip_check_sex_and_relatedness': False,


Was this just a stray line?

The tests were failing because that line was missing. No idea why but it didn't seem related to my changes.

Interesting, I will take a look into this as it is unexpected!

bpblanken · 2025-08-05T16:53:46Z

v03_pipeline/lib/paths.py

@@ -1,6 +1,8 @@
 import hashlib
 import os

+from uuid import uuid1


I think it might actually be better to use the request_id (currently a timestamp) being assigned by the pipeline_worker to instead be passed in to loading_pipeline_queue_path(). That value is lexically sortable, for example:

20250805-123456 20250805-223000 20250806-001500

Good idea! I'll give it a go

I've ultimately decided to do a combination of both methods. A queue file would look something like this: request_20250805-123456_<uuid>.json. I chose to keep the uuid in there to be prevent losses in jobs when multiple jobs are submitted within the same second

bpblanken · 2025-08-05T16:58:23Z

v03_pipeline/lib/paths.py

+        f'request_{uuid1().int}.json',
+    )
+
+def get_oldest_queue_path() -> str:


This method might actually fit nicely in runs.py. You'll probably want to abstract the

os.path.join(LOCAL_DISK_MOUNT_PATH, 'loading_pipeline_queue')

into a helper in some way that would be shared between loading_pipeline_queue_path(run_id: str) and this method!

I moved the method to runs.py :)

bpblanken · 2025-08-05T17:01:10Z

@nvnieuwk Thanks for contributing! This has been on our roadmap and is definitely needed. I left a few comments, and we'll need to get the build passing, but the idea looks solid to me.

bpblanken · 2025-08-05T17:13:49Z

v03_pipeline/api/app.py

@@ -35,24 +35,6 @@ async def loading_pipeline_enqueue(request: web.Request) -> web.Response:
    except ValueError as e:
        raise web.HTTPBadRequest from e

-    try:


we don't need to do this on this pr, but a back pressure notion of "there are too many files in the queue" would make this more complete! I can add a ticket.

Good idea, what would be 'too many files' in this case?

what about 5 or 10?

We would probably want to queue more in our case. Would it be bad if the limit is higher? (I'm thinking 1000 even)

We could make it an env var? The reason for the low suggested limit is just the simplicity of the mechanism. If you're reliably queuing up dozens of requests from the UI, you might want to consider joint calling your VCF and loading multiple projects at once.

We're actively trying to make the pipeline more performant and better able to support concurrent loads as well, which will hopefully ease some of this burden!

That's a good idea, thank you! We sadly can't do analysis on large cohorts at our lab due to several reasons, but having a way to configure that limit might be the best way to handle this.

Also really looking forward to the concurrent loads!

I added the LOADING_QUEUE_LIMIT environment variable with a default of 10 for this. The app will first check if the queue is full before adding a file to it and will return a 409 error in this case.

bpblanken · 2025-08-06T15:09:39Z

v03_pipeline/lib/paths.py

+    )
+    return os.path.join(
+        loading_pipeline_queue_dir(),
+        f'request_{run_id}_{uuid.uuid1().int}.json',


It might be simpler to just use random.randint(0, 99) or something similar here. We run into issues with run_ids being too big in our system sometimes, so just keeping it less verbose would be nice!

Ah, I see that you're not including it when you regex parse downstream. I think if you shorten the randomness you should be fine to include it as part of the run id.

Good idea, I'll just use the first 5 characters or something like that

bpblanken

@nvnieuwk thanks for incorporating the feedback! looks like I might need to move a few imports around to get the build passing but things looks fully functional and correct to me!

nvnieuwk · 2025-08-07T15:14:16Z

Awesome! I can also take a look at it tomorrow if you haven't done it by then :)

nvnieuwk added 3 commits August 5, 2025 17:09

allow multiple entries in the queue

78ef09e

fix an issue with empty dirs

0c6cbf4

change function name

b326688

nvnieuwk requested a review from a team as a code owner August 5, 2025 15:28

bpblanken reviewed Aug 5, 2025

View reviewed changes

nvnieuwk added 2 commits August 6, 2025 13:13

use a timestamp as identifier

f59811c

move get_oldest_queue_path to runs.py

7bcce18

bpblanken reviewed Aug 6, 2025

View reviewed changes

nvnieuwk added 2 commits August 7, 2025 11:29

add a queue limiter with env variable

9fca024

shorten the resource filename and update the run ID

77a4ad5

bpblanken approved these changes Aug 7, 2025

View reviewed changes

bpblanken merged commit 7d6278f into broadinstitute:main Aug 7, 2025
1 check failed

nvnieuwk deleted the real-queue branch August 8, 2025 07:38

Convert the v03 pipeline queue to a real queue #1136

Convert the v03 pipeline queue to a real queue #1136

Uh oh!

Conversation

nvnieuwk commented Aug 5, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bpblanken commented Aug 5, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nvnieuwk Aug 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bpblanken left a comment

Choose a reason for hiding this comment

Uh oh!

nvnieuwk commented Aug 7, 2025

Uh oh!

Uh oh!

Uh oh!

nvnieuwk Aug 7, 2025 •

edited

Loading