Pattern for Flows that contain Child Jobs with multiple Parent Jobs

[hhopkins95]

Hello --

Is there a recommended pattern to accomplish flows in which multiple jobs need to await the same child job?

For example, something similar to :

where the completion of "Job A" should trigger execution of jobs B, C, and E?

I have tried defining the flow with this approach :

const jobA : FlowChildJob = { 
name : "jobA", 
queueName : "jobAQueue", 
data : {...data}, 
opts : { 
jobId : "someUniqueId"
}}

const jobB : FlowChildJob = { 
...rest, 
opts : { 
jobId : "someUniqueId"
}, 
children : [jobA]
}

const jobC : FlowChildJob = { 
...rest, 
opts : { 
jobId : "someUniqueId"
}, 
children : [jobA]
}

const jobD : FlowChildJob = { 
...rest, 
opts : { 
jobId : "someUniqueId"
}, 
children : [jobB, jobC]
}

const jobE : FlowChildJob = { 
...rest, 
opts : { 
jobId : "someUniqueId"
}, 
children : [jobA]
}

const flowJob : FlowJob = { 
...flowJobParams, 
children : [jobE, jobD]
}

though, the child job will only trigger the parent who lists it as a child last -- (Job A will only trigger Job E, in the example above -- not B or C).

I'm considering creating a "CopyCat" queue like this :

const jobA : FlowChildJob = { 
...rest,
queueName : "jobAQueue", 
opts : { 
jobId : "someUniqueId"
}}

const jobACopycat1 : FlowChild = { 
queueName : "CopycatQueue". 
data : { 
copyJobId : jobA.opts.jobId
}

const jobB : FlowChildJob = { 
...rest,
children : [jobA]
}

const jobC : FlowChildJob = { 
... rest,
children : [jobACopycat1]
}


// ...rest of jobs

and a Copycat processor :

type CopycatJob = { 
copyJobId : string
}

export const copycatProcessor = async (job : Job<CopycatJob>) => { 
const {copyJobId} = job.data 
while (true) { 
// read the "copyJob" 
// copy statuses / data to this job
// wait some interval 
// break when necessary
}
return 
} 

Where the jobId for the "base" (Job A in this example) job will be passed as data to the CopyCat job(s) (JobA copy 1, JobA copy2).

The copycat worker will just read it's "base" job on some interval and copy over the statuses / results when the job completes.

This way every job will only have a single parent.

Is this a good solution, or are there cleaner ways to handle these types of flows?

I'm not sure if I'm overlooking anything that could be an issue with this implementation.

Thanks

[hhopkins95]

After some initial testing, the method above seems to work --

For anybody looking for a similar solution, here is some code that may help :

import { Job, Queue } from "bullmq";
import { wait } from "../../lib/utils/typescript/wait";
import { QUEUE_NAMES } from "../__QUEUE_CONFIGS/queue-names";
import { getRedisConnectionParams } from "../_helpers/getRedisConnectionParams";

export type CopyCatJob = {
  copyJobId: string;
  copyJobQueueName: keyof typeof QUEUE_NAMES;
};

export const copy_cat_processor = async (job: Job<CopyCatJob>) => {
  try {
    const copyJobId = job.data.copyJobId;
    const copyJobQueueName = job.data.copyJobQueueName;

    if (!copyJobId) throw new Error("copyJobId is required");
    if (!copyJobQueueName) throw new Error("copyJobQueueName is required");
    if (copyJobQueueName === "copy_cat")
      throw new Error("Cannot create a copy job for the copy_cat queue");

    const copyQueue = new Queue(copyJobQueueName, {
      connection: getRedisConnectionParams(),
    }); // create ref for the copy queue

    await wait(100); // I wait for a moment to make sure the base job is created in it's queue -- maybe a better way to do this, but it fits the needs of my project
    let jobMissingCount = 0 // just in case, I check a few times if the base job is missing before throwing an error

    while (true) {
      // check the status of the copy job
      const copyJob = await copyQueue.getJob(copyJobId);

      // if the copy job is missing from it's queue, try again
      if (!copyJob) {
        jobMissingCount++;
        if (jobMissingCount > 3) {
          throw new Error(
            `Job to copy not found after 3 attempts. JobId: ${copyJobId} Queue: ${copyJobQueueName})`,
          );
        }
        await wait(300);
        continue;
      }

      // Check status of the copy job
      const status = await copyJob.getState();
      if (status == "completed") {
        const copyJobRes = copyJob.returnvalue;
        return copyJobRes;
      }

      if (status == "failed") {
        throw new Error(`Copy job failed: ${copyJobId}`);
      }

      await wait(3000); // wait 3 seconds before checking on the job again
    }
  } catch (error) {
    throw new Error(`Error CopyCat processor: ${error}`);
  }
};

a helper function to create copies of a job :

import { FlowChildJob } from "bullmq";
import { QUEUE_NAMES } from "../__QUEUE_CONFIGS/queue-names";
import { JOB_CONFIGS } from "../__QUEUE_CONFIGS/job-configs";
import { CopyCatJob } from "../job-processors/__COPYCAT.processor";
import { v4 as uuid } from "uuid";

export const createCopyCatJob = (job: FlowChildJob): FlowChildJob => {
  const baseQueueName = job.queueName as keyof typeof QUEUE_NAMES;
  const baseJobId = job.opts?.jobId;

  if (!baseJobId) throw new Error("Base Copy Cat job must have a jobId");

  const jobData: CopyCatJob = {
    copyJobId: baseJobId,
    copyJobQueueName: baseQueueName,
  };

  const copyCatJob: FlowChildJob = {
    name: `copy_cat_job_for_${baseJobId}`,
    queueName: QUEUE_NAMES.copy_cat,
    data: jobData,
    opts: {
      ...JOB_CONFIGS.copy_cat,
      jobId: `${baseJobId}-copy_cat-${uuid()}`, // add a uuid in case there are multiple copy cat jobs for the same base job
    },
  };

  return copyCatJob;
};

example usage :

import { FlowChildJob, FlowJob } from "bullmq";

const createFlowJob = (flowId : string) => { 

const jobA : FlowChildJob = { 
name : "JobA", 
queueName : "JobAQueue", 
}

// create copy cat jobs 
const jobA_copy1 = createCopyCatJob(jobA)
const jobA_copy2 = createCopyCatJob(jobA)


const jobB : FlowChildJob = { 
name : "JobB", 
queueName : "JobBQueue", 
children : [jobA]
}

const jobC : FlowChildJob = { 
name : "JobC", 
queueName : "JobCQueue", 
children : [jobA_copy1]
}


const jobD : FlowChildJob = { 
name : "JobD", 
queueName : "JobDQueue", 
children : [jobA_copy2]
}



const flowJob : FlowJob = { 
name : "flowjob", 
queueName : "flowQueue", 
children : [jobB, jobC, jobD]
}

A couple considerations about this implementation that may be relevant if you choose to use it :

All copy cat jobs will immediately be added to the active state, regardless where their base job lives in the flow tree. So, if the base job relies on other long running processes before beginning execution, the copy cat could be running in the background for a while.

I accounted for this by setting the global concurrency on the copy cat worker relatively high just to make sure it will be active whenever it's base worker is doing it's work (not waiting on other copy cats listening to other queues that could be delayed to clear the copy queue first).

Also -- I set the worker configs on the queues that are being "copied" to remove completed / failed jobs based on time, not number. If you ever run into a scenario where your queues are turning over jobs faster than the interval at which the copy_cat is checking, it's possible for the copy job to have been deleted before the copy_cat can get back around to checking again.

Pretty unlikely scenario, but just thought I'd mention it

[manuganji]

@hhopkins95 You probably have removeOnComplete set to true? It is probably the reason for the parents not getting triggered by the completion of the child job?

[manast]

removeOnComplete should not affect triggering the parent job.

[CodingMeSwiftly]

It would be really nice to have multiple parents work natively in flows without the need for a workaround like that.
This feature seems to be at the root of many requests/issues: #2087, #1163, #1021

[fakhir]

Another related request is #2780.