Add ability to wait for job to complete

[kerwanp]

Problem

There is currently no way to dispatch a job and wait for its result: "Fire and forget". Once a job is dispatched, the caller only receives a { jobId } and has no built-in mechanism to know when the job completes or to retrieve the resulted value.

This is a common need when a job performs work that the caller depends on. For example, generating a PDF, sending an email and getting a delivery status, or offloading a heavy computation to a worker and awaiting the outcome.

This also open the possibility for building "workflow jobs":

class DocumentProcessingJob extends Job {
  async execute() {
    const document = await Document.findOrFail(this.payload.documentId)

   const content = await ParseMediaJob.dispatch({ documentId: document.id }).wait()


   const analysis = await AnalyzeDocumentJob.dispatch({ parsed })

   const [embedding, investments, parcel, commitments] = await Promise.all([
     VectorizeDocumentJob.dispatch({ content }).wait(),
     ExtractDocumentInvestmentJob.dispatch({ content }).wait(),
     ExtractDocumentParcelJob.dispatch({ content }).wait(),
     ExtractDocumentCommitmentsJob.dispatch({ content }).wait(),
   ])

   await document.update({
     embedding,
     investments,
     commitments,
     parcel,
   })
  }
}

Use Cases

Benefit from job orchestration

Currently jobs are only "fire and forget", while this is great for sending notifications, processing data, etc.
In some scenarios we want to benefit from retry mechanism, offload computation, etc but still expect a result.

It is a common practice to communicate with third-party services using workers to reduce impact if the external API has ongoing issues by benefiting from replay mechanism.

Off-load computation

When doing work that require heavy computation or specific infrastructure components (e.g servers with GPU) we want to return the result at completion without having to configure a complete pub/sub system for real-time.

Proposed API

const result = await SendEmailJob.dispatch(payload).wait(timeoutMs)

Design considerations

Extending dispatchers

Dispatchers would now have a wait method that would first dispatch the job(s) and wait for the result(s). The promise would be resolved once the job is completed.

Storing the result

As of now, execute() returns Promise<void>. It would need to support returning a value, and the worker completion path would need to serialize and persist that value. This means:

• Adding a result field to JobRecord
• The completeJob adapter method must accept an optional result and store inside JobRecord

Notification mechanism

There are several possible implementations for waiting job completion:

Polling

After dispatching with wait() it would poll the Adapter#getJob method until the status is completed or failed.
This would work with every adapter out of the box. Simple but introduces latency proportional to the poll interval and wastes resources under loads.

Polling could be optimized by havin a global poll instead of per job awaited

Pub/Sub

By using real-time notification channel (depends on the adapter) the worker would publish the result of the job and the client subscribe.
This would be near-instant and could even be used in the other direction (to have instant dispatch). This it is adapter-specific and for Knex adapter it would require to fallback to polling if not on Postgres.

I've done a prototype of a library for building micro-services using this approach (Redis, AMQP, TCP)

The result could still be persisted so monitoring solutions can access that information.

[kerwanp]

@RomainLanz I started a draft PR available here. As discussed it uses polling for fetching the job result.

Currently I am focusing on the knex adapter and will update other adapters accordingly once we have settled on the implementation.

Design implications

Job output persistence

Option A: Adding output to JobRecord, this imply creating a new column on queue_jobs
Option B: Adding output to JobData, this imply retrieving the record, deserializing job data and re-serializing it in completeJob

I opted for the Option A as job completion only require a simple UPDATE but this would introduce a breaking change.

Keep job on completion

The default job retention configuration removeOnComplete and removeOnFail is set to undefined for both. (remove by default)
This behavior cannot work with polling as the persisted job record is removed on job completion.

In my opinion this should be set to true or to a defined value (e.g 30 days) by default for the following reasons:

• This is the common behavior (see Laraval Queue, BullMQ, etc)
• This allow monitoring tools to work by default
• A miss-informed user could skip configuring it and realize too late that it is not configured
• It makes polling for job completion not work out of the box