[RFC] queue_job: enqueue dependent job in jobrunner #884

Activity

1. guewen commented on Jan 23, 2026

   

   guewen

   on Jan 23, 2026

   Member

   More actions

   It makes sense for the reasons you mention.
   How would the jobrunner identify waiting jobs ready to process, is there any other way than polling?
   I'm wondering if we could leverage NOTIFY payloads from the worker that ends a job, so the jobrunner could check only a subset of the jobs.
2. sbidoul commented on Jan 23, 2026

   

   sbidoul

   on Jan 23, 2026

   MemberAuthor

   More actions

   The notification payload already has the job uuid.

   But thinking about it, it means if the jobrunner misses a done notification dependencies will stay waiting. Hm...
3. guewen commented on Jan 26, 2026

   

   guewen

   on Jan 26, 2026

   Last edited by guewen

   Member

   More actions

   Not 100% sure it is correct, but this query should allow to find the jobs in wait_dependencies, it could be in the jobrunner while loop.

   explain (analyze, verbose)
   SELECT queue_job.id, queue_job.uuid
   FROM queue_job
   WHERE queue_job.state = 'wait_dependencies'
   AND NOT EXISTS (
     SELECT 1
     FROM json_array_elements_text(queue_job.dependencies::json->'depends_on') AS parent_uuid
     JOIN queue_job parent_job
       ON parent_job.graph_uuid = queue_job.graph_uuid 
       AND parent_job.uuid = parent_uuid
     WHERE parent_job.state != 'done'
   );

   However, the query plan is not good enough (here I have ~5k jobs in wait_dependencies).

                                                                                        QUERY PLAN                                                                                      
   -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
    Index Scan using queue_job__state_index on public.queue_job  (cost=0.43..156305.69 rows=1170 width=41) (actual time=15.904..925.628 rows=4 loops=1)
      Output: queue_job.id, queue_job.uuid
      Index Cond: ((queue_job.state)::text = 'wait_dependencies'::text)
      Filter: (NOT (SubPlan 1))
      Rows Removed by Filter: 3229
      SubPlan 1
        ->  Hash Join  (cost=64.81..66.19 rows=1 width=0) (actual time=0.285..0.285 rows=1 loops=3233)
              Hash Cond: (parent_uuid.value = (parent_job.uuid)::text)
              ->  Function Scan on pg_catalog.json_array_elements_text parent_uuid  (cost=0.01..1.01 rows=100 width=32) (actual time=0.002..0.003 rows=4 loops=3233)
                    Output: parent_uuid.value
                    Function Call: json_array_elements_text(((queue_job.dependencies)::json -> 'depends_on'::text))
              ->  Hash  (cost=64.65..64.65 rows=12 width=37) (actual time=0.281..0.281 rows=776 loops=3233)
                    Output: parent_job.uuid
                    Buckets: 2048 (originally 1024)  Batches: 1 (originally 1)  Memory Usage: 116kB
                    ->  Index Scan using queue_job__graph_uuid_index on public.queue_job parent_job  (cost=0.43..64.65 rows=12 width=37) (actual time=0.013..0.188 rows=776 loops=3233)
                          Output: parent_job.uuid
                          Index Cond: ((parent_job.graph_uuid)::text = (queue_job.graph_uuid)::text)
                          Filter: ((parent_job.state)::text <> 'done'::text)
                          Rows Removed by Filter: 42
    Query Identifier: -8629826990232030693
    Planning Time: 0.150 ms
    Execution Time: 925.659 ms

   It would probably be better with a jsonb and an index directly on the depends_on key.

   Perhaps, when do both

   • When a job is done: for each uuid in reverse_depends_on a NOTIFY is sent with the uuid, and ChannelManager.notify now check if it is or not ready to be executed according to the dependencies
   • A catch-up is added in the loop in case of misses
     Maybe the notify is not needed at all

   Maybe the first step would be to check if we can have a faster query that finds resolved dependencies using jsonb or another execution plan. Also maybe an index on state = 'wait_dependencies'
4. guewen commented on Jan 27, 2026

   

   guewen

   on Jan 27, 2026

   Last edited by guewen

   Member

   More actions

   Already better by removing the join on graph_uuid actually (but not so good)

   explain (analyze, verbose)
   SELECT queue_job.id, queue_job.uuid
   FROM queue_job
   WHERE queue_job.state = 'wait_dependencies'
   AND NOT EXISTS (
     SELECT 1
     FROM json_array_elements_text(queue_job.dependencies::json->'depends_on') AS parent_uuid
     JOIN queue_job parent_job
       ON parent_job.uuid = parent_uuid
     WHERE parent_job.state != 'done'
   );
   
    Index Scan using queue_job__state_index on public.queue_job  (cost=0.43..18944.68 rows=667 width=41) (actual time=5.629..36.088 rows=4 loops=1)
      Output: queue_job.id, queue_job.uuid
      Index Cond: ((queue_job.state)::text = 'wait_dependencies'::text)
      Filter: (NOT (SubPlan 1))
      Rows Removed by Filter: 1179
      SubPlan 1
        ->  Nested Loop  (cost=0.56..259.26 rows=20 width=0) (actual time=0.030..0.030 rows=1 loops=1183)
              ->  Function Scan on pg_catalog.json_array_elements_text parent_uuid  (cost=0.01..1.01 rows=100 width=32) (actual time=0.003..0.004 rows=10 loops=1183)
                    Output: parent_uuid.value
                    Function Call: json_array_elements_text(((queue_job.dependencies)::json -> 'depends_on'::text))
              ->  Index Scan using queue_job__uuid_index on public.queue_job parent_job  (cost=0.55..2.57 rows=1 width=37) (actual time=0.002..0.002 rows=0 loops=11782)
                    Output: parent_job.id, parent_job.date_done, parent_job.func_string, parent_job.max_retries, parent_job.result, parent_job.exc_info, parent_job.retry, parent_job.user_id, parent_job.uuid, parent_job.name, parent_job.priority, parent_job.state, parent_job.eta, parent_job.date_enqueued, parent_job.date_created, parent_job.date_started, parent_job.model_name, parent_job.company_id, parent_job.job_function_id, parent_job.channel, parent_job.identity_key, parent_job.method_name, parent_job.args, parent_job.kwargs, parent_job.channel_method_name, parent_job.job_batch_id, parent_job.records, parent_job.exec_time, parent_job.exc_name, parent_job.exc_message, parent_job.worker_pid, parent_job.date_cancelled, parent_job.graph_uuid, parent_job.dependencies
                    Index Cond: ((parent_job.uuid)::text = parent_uuid.value)
                    Filter: ((parent_job.state)::text <> 'done'::text)
                    Rows Removed by Filter: 0
    Query Identifier: 15504662281462197
    Planning Time: 0.172 ms
    Execution Time: 36.127 ms
5. github-actions commented on Aug 2, 2026

   

   github-actionsBot

   on Aug 2, 2026 – with GitHub Actions

   More actions

   There hasn't been any activity on this issue in the past 6 months, so it has been marked as stale and it will be closed automatically if no further activity occurs in the next 30 days.
   If you want this issue to never become stale, please ask a PSC member to apply the "no stale" label.
6. github-actions

   added

   stalePR/Issue without recent activity, it'll be soon closed automatically.PR/Issue without recent activity, it'll be soon closed automatically.

   on Aug 2, 2026
7. github-actions

   closed this as not plannedon Sep 6, 2026
8. sbidoul

   reopened this on Sep 8, 2026
9. github-actions

   removed

   stalePR/Issue without recent activity, it'll be soon closed automatically.PR/Issue without recent activity, it'll be soon closed automatically.

   on Sep 13, 2026