Add support for unique jobs

@SmovS please post an interdiff when posting patches, or please make a commit to an existing MR. If you do this, we can see the proposed change to the patch; in this case it would make it easier to see the logging code you have removed.

+++ b/src/Plugin/AdvancedQueue/Backend/Database.php
@@ -2,22 +2,32 @@
+use Drupal\Core\Logger\LoggerChannelFactoryInterface;

@@ -26,6 +36,13 @@ class Database extends BackendBase implements SupportsDeletingJobsInterface, Sup
+  /**
+   * The logger factory.
+   *
+   * @var \Drupal\Core\Logger\LoggerChannelFactoryInterface
+   */
+  protected $loggerFactory;
+

@@ -37,13 +54,18 @@ class Database extends BackendBase implements SupportsDeletingJobsInterface, Sup
+   * @param \Drupal\Core\Logger\LoggerChannelFactoryInterface $logger_factory
+   *   The logger factory.
...
+    $this->loggerFactory = $logger_factory;

@@ -55,7 +77,9 @@ class Database extends BackendBase implements SupportsDeletingJobsInterface, Sup
+      $container->get('logger.factory')

All this logger code can be removed too as nothing is using $this->logger AFAICS.

Added a review to the MR

Latest changes by @smovs but as a patch file for those who need it.

> public $ensure_unique = self::UNIQUE_OVERWRITE;

I think the default value for this should be UNIQUE_DONT_OVERWRITE, otherwise behaviour will be changed for existing job type plugins.

Unit tests are currently broken because the MR introduces a dependency on the kernel to the Job class:

  public function toArray() {
SNIP
      'unique_id' => $this->getUniqueId(),
  }

// and then:

  public function getUniqueId() {
    if (!\Drupal::service('plugin.manager.advancedqueue_job_type')->jobTypeRequiresUniqueness($this->type)) {
SNIP
  }

Not sure how best to resolve this. Though I am not really keen on toArray() calling getUniqueId() to get the value rather than just returning what is set in the class. The order of precedence in getUniqueId / setUniqueId / constructor for handling $this->uniqueId doesn't seem right to me.

Another issue: we're using the phrase 'unique ID' in comments and method names and var names.

But there is *already* a unique ID -- the Job ID $job->id which is a unique ID. And it's fairly common in core to casually prefix the word 'ID' with 'unique'.

So I think it's going to be really confusing to differentiate between 'the job ID, which is a unique ID', and the 'unique ID which is a hash to check for uniqueness'.

I'd really like to rename the constants for these, as they're confusing:

  /**
   * Allow duplicate jobs.
   */
  const UNIQUE_DUPLICATES = 'allow'; // Doesn't say what happens to duplicates!

  /**
   * Do not enqueue the new job.
   */
  const UNIQUE_DONT_OVERWRITE = 'leave'; // 'don't overwrite' could mean the duplicate is written to a new job, which is the 'allow' case!

  /**
   * Overwrite existing job.
   */
  const UNIQUE_OVERWRITE = 'overwrite';

Something like:

  const UNIQUE_ALLOW_DUPLICATES = 'allow';
  const UNIQUE_SKIP_DUPLICATES = 'leave';
  const UNIQUE_OVERWRITE_DUPLICATES = 'overwrite';

Furthermore, what about making them an Enum now that we have those in PHP?

I'm also a bit confused about the need for BackendBase::ensureUniqueJobsSupport(). This says:

   * Backends that do not support unique jobs should call this from
   * enqueueJobs() to ensure that jobs that require uniqueness are not queued.

But the Queue class's enqueueJob/s() methods already have a check.

And the module's README has this code example showing that jobs should be queued by calling these:

$queue = Queue::load('default');
$queue->enqueueJob($job);

However! Kernel tests, both in this MR and the main branch, repeatedly do this:

    $this->firstQueue->getBackend()->enqueueJobs([$first_job, $third_job]);

Again, not sure how best to resolve it as it's a bit of a confusing situation.

Everyone.. I truly dislike issues going over 100 comments, and this one is really in danger of that.. Can I ask those that are interested in this to review and test this, please??

@joachim Can you please update the issue summary to explain what is the current status of this, and if there's still any blocker / caveat that would prevent this to be marked RTBC by someone else?

Setting this to "Needs review" as I'm not sure the "Needs work" that was added back in #69 is still applicable.

I reviewed the code and have no objections in merging it back. I also tested the new functionality in a project where I need the functionality and it works flawlessly.

Mark as RTBC

I think this needs work, as per my various comments -- see #93 and #94.

Regarding your comments:

I think its safe to remove the comment, as enqueueJob and enqueueJobs will throw an exception if you try to add a JobItem that needs uniqueness and the queue does not support it.

Regarding the second comment. The Queue itself provides two function enqueueJob and enqueueJobs to add jobs, which looks fine to me.

Oh and #92 as well.

@joachim Oh yes, let's do this. I had my problems understanding them as well.

OK. Then we need to work on #92 and removing the first comment of #93?

stmh → changed the visibility of the branch advancedqueue-2918866-2918866-add-support-for to hidden.

Worked on #92 and #93, see my latest commits.

Pushed a bunch of fixes. Tests now pass.

Still to do:

1. My comment in #93. I'm not sure how best to resolve this and it would be useful to have input from the original author of that code. But it looks like the checking in Queue was added in #48 back in 2019, and the checking in BackendBase even before that, so I don't know if those devs are even still around. Given that, I'd be inclined to remove the checks in Queue and let backends use the ensureUniqueJobsSupport() helper.

2. getDuplicateJobId could probably be made protected. That would make SupportsUniqueJobsInterface an empty interface, but there's already precedent for that in the backend interfaces.

3. I REALLY want to rename the DB column we're adding from unique_id to something like job_hash. It's going to be a source of confusion and bugs with the current name. I appreciate that sites are already using this patch.... but that's a risk you take when using a patch.

4. The MR makes changes to existing hook_update_N() implementations, which is out of scope AND generally not a good idea:

-  $schema = Database::getConnection()->schema();
-  $schema->addIndex('advancedqueue', 'queue_state', ['state'], $spec['advancedqueue']);
+  $database_schema = \Drupal::database()->schema();
+  $database_schema->addIndex('advancedqueue', 'queue_state', ['state'], $spec['advancedqueue']);

These were first introduced in #51 -- and later patches have added the same changes to newer implementations that were added to the module in the meantime!

3. I REALLY want to rename the DB column we're adding from unique_id to something like job_hash. It's going to be a source of confusion and bugs with the current name. I appreciate that sites are already using this patch.... but that's a risk you take when using a patch.

FWIW I've been using "fingerprint" as the column name for this purpose in a few other places. It is fairly common in various libraries.

I do agree that "job_hash" sounds better than "unique_id" at least. Every ID is unique so the column name sounds confusing.

Just wanted to comment that I think this should not be committed to the module before we release 1.0, as adding new features is somewhat against what we should be doing in "rc" status.

Then again, we should probably release 1.0 ASAP unless someone has any objections, and if there is any, please comment on 🌱 Roadmap to 1.0 Active .

1.0 has now been released. This can easily go into 1.1, we just need to get this in RTBC.

The patch with latest updates for those who are not comfortable downloading merge requests.

I think part of the problem we're having with naming things and figuring out how this should be organised is that the semantics of 'unique' are unhelpful.

Let's discard the whole term "unique" and focus on the language of duplicates and how we handle duplicates: we need to detect duplicates (by hash), and then manage them.

Concretely this leads me to the following suggestions:
instead of a single Drupal\advancedqueue\Plugin\AdvancedQueue\Backend\SupportsUniqueJobsInterface we have:
Drupal\advancedqueue\Plugin\AdvancedQueue\Backend\SupportsDetectingDuplicateJobsInterface

Our annotations become:

/**
   * Allow duplicate jobs works with any backend.
   */
  public const DUPLICATES_ALLOW = 'allow';

  /**
   * Do not enqueue duplicate jobs. Requires a backend implementing SupportsDetectingDuplicateJobsInterface.
   */
  public const DUPLICATES_SKIP = 'skip';

  /**
   * Merge with existing job. Requires a backend implementing SupportsDetectingDuplicateJobsInterface and SupportsDeletingJobsInterface.
   */
  public const DUPLICATES_MERGE = 'merge';

Notice that DUPLICATES_MERGE I've said requires SupportsDeletingJobsInterface instead of some new SupportsUpdatingJobsInterface - this is because the simplest possible way of updating a job is to delete the old one and insert a new one.

I'm saying 'merge' here instead of the previous 'overwrite' because the basic idea in order to avoid committing us to what the payload should be of the final updated job; in this first implementation we should just overwrite the old job with the new one, but in future we might allow the job type plugin to have an opinion about how the payload of old and new job are combined.

I agree that BackendBase::ensureUniqueJobsSupport() does not seem necessary. It's unused in the Database backend which is the only one we offer.

I agree SupportsUniqueJobsInterface doesn't need a public getDuplicateJobId() method.

I dislike the jobTypeRequiresUniqueness() and getJobTypeUniquenessType() methods on the JobTypeManager. I understand the logic for wanting to check this without instantiating a plugin, but the semantics seems very ugly. I think it would be nicer to have this as just a single method on the Job class: Job::getDuplicatesPolicy() which itself calls out to JobTypeManager::getDefinition().

I agree $this->getBackend()->supportsJobType($job_type); is a nicer semantic than trying to figure this out on the Queue entity or in the JobTypeManager. Actually if we do this, then strictly speaking we don't even need the SupportsUnique/Duplicates interface at all.

I've opened a new MR as I'm making a significant number of changes here:
- using 'duplicates' as the key term instead of 'unique'
- introducing the term 'duplicates policy' to describe the duplicates handling strategy specified on a job type.
- using 'fingerprint' instead of 'unique id' or 'job hash'
- removing jobTypeRequiresUniqueness() and getJobTypeUniquenessType() methods from the JobTypeManager, in favour of new method Job::getDuplicatesPolicy()
- removing SupportsUniqueJobsInterface altogether, adding no new interface
- adding a verifyJobSupport() method to BackendInterface and BackendBase.

However, doing this I've realised that a lot of the reason why this issues is getting tricky goes back to #20:

Backends may or may not support unique jobs.

However, silently ignoring jobs that should be unique is dangerous, ... Commerce Recurring's jobs absolutely MUST NOT be used with a backend that does not support unique jobs, and therefore, I think we should guard against this here.

So therefore, I want to introduce a concept of a job type *requiring* that its jobs be treated as unique. This can be declared on the job type plugin annotation.

This goal makes sense. However, it is not really acheivable for the reasons pointed to in #93. It is perfectly possible and reasonable for custom code using this module to enqueue jobs with:
$queue->getBackend()->enqueueJobs([$jobs]);
This means that calling code has direct access to the backend plugins. We can't stop calling code from passing jobs to backend plugins that can't properly handle them, and we can't force custom or contrib backend plugins to verify they can handle a job properly.

Everything we are doing with SupportsUniqueJobsInterface, Queue::verifyQueueSatisfiesJob(), BackendInterface::verifyJobSupport(), JobTypeManager::jobTypeRequiresUniqueness() etc is basically wishful thinking. We are trying to create protective mechanisms but we can't ensure that anyone uses them.

I think at this point we should:
- rip out these mechanisms (which in MR11 are basically simplified to BackendInterface::verifyJobSupport())
- create a new issue for discussing and working on this rather tricky problem, which may apply to other future features not just this duplicates feature
- document on the annotation type that duplicates handling may not be safe when using backend other than the Database backend.

> Let's discard the whole term "unique" and focus on the language of duplicates and how we handle duplicates: we need to detect duplicates (by hash), and then manage them.

I sort of see what you're getting at, but the word 'duplicate' carries the same connotations as 'unique' -- ensuring there's only one of something.

> I think it would be nicer to have this as just a single method on the Job class: Job::getDuplicatesPolicy() which itself calls out to JobTypeManager::getDefinition().
> - removing jobTypeRequiresUniqueness() and getJobTypeUniquenessType() methods from the JobTypeManager, in favour of new method Job::getDuplicatesPolicy()

The Job class can't depend on the service container!!!! I *removed* that in my latest commits. It crashes the unit tests otherwise.

> Notice that DUPLICATES_MERGE I've said requires SupportsDeletingJobsInterface instead of some new SupportsUpdatingJobsInterface - this is because the simplest possible way of updating a job is to delete the old one and insert a new one.

Then it's not 'merge', it's 'replace'. Saying 'merge' when it's actually deleting the old one is misleading.

Is there enough value in actually support the "merge" strategy? Or the concept of strategies at all?

In all other implementations I've used, a job/task is either unique/deduplicated (meaning that a job with the same fingerprint is rejected on enqueue) or it isn't.

My assumption without reading the code would be that I do:

// If I don't provide a fingerprint, it is autogenerated for me.
$job = Job::createUnique("my_type", $payload, ['fingerprint' => 'my-unique-value']);

enqueueJobs() can check if $job->getFingerprint() is non-empty, and if the backend does not implement SupportsUniqueJobsInterface, throw an exception.

I tend to support #113 as well. From the point of view of maintenance, it makes the code a lot simpler, and I think also from the point of view of an API user. As an API user, I'd try to create the job, and if I get the exception, I either propagate the error up, or I do something to make the job unique and try again. Normally, the API user is the one that would better understand the consequences of trying to create duplicate jobs.

Thanks to everyone for the feedback.

What I've done is:
- simplified the job type annotation to a boolean
- given the job type plugin responsibility for deciding how duplicates should be handled
- made the base job type plugin's implementation simply throw an exception if there's a duplicate
- given the Queue entity, not the Database backend, responsibility for looking for duplicates.

This means that all of the duplicates functionality added in this issue is completely bypassed if you call $queue->getBackend()->enqueue(). But if you call $queue->enqueue() you get safe behavior whatever backend you use.

I believe in a follow-up issue we should deprecate Queue::getBackend(). This function creates a leaky abstraction of sorts; by putting API users in direct contact with the backend, we lose any ability to intermediate. Instead API users should interact with the queue entity, and we can use that to provide safety and backwards compatability as we add new features like this in future.

I like where this has got too. It seems clean and simple, but flexible enough to accomodate any use case.

Green at last.

@jonathanshaw, there's a typo in Merge request 11 on lines 132-133 of advancedqueue.install. The variable initialized is $schema but undefined $database_schema name is used. See attached patch.

Thanks @adamfranco, I fixed the variable name in the update hook.

Hi @jonathanjfshaw,

thanks for your work and your MR.

I tested your changes by:
- creating a new AdvancedQueueJobType plugin with allow_duplicates = false configuration;
- added a code to create a new job after submitting a custom form.

as a result after creating the second job with the same value, I got a fatal error - Drupal\advancedqueue\Exception\DuplicateJobException: Job rejected because it duplicates existing jobs.

I guess, we should simply not add the second job to the queue and add a log with a notice.

What do you think?

I got a fatal error - Drupal\advancedqueue\Exception\DuplicateJobException: Job rejected because it duplicates existing jobs.

I guess, we should simply not add the second job to the queue and add a log with a notice.

Yes, that behavior is by design. This strategy was chosen by the maintainer in #114:

As an API user, I'd try to create the job, and if I get the exception, I either propagate the error up, or I do something to make the job unique and try again. Normally, the API user is the one that would better understand the consequences of trying to create duplicate jobs.

yeah, it works for me with try-catch.

I guess it should be somehow documented on the module page / readme.

Patch with the latest changes from MR to be used in 🐛 multiple recurring orders get created if cron is badly configured Needs work

Updated patch for 1.0 version instead of 1.x-dev branch

jcnventura → closed merge request !9

jcnventura → closed merge request !11

Thanks everyone! And there were a lot of you. As to how to document, this.. I guess it will be part of the release notes when the next release is created. If someone would be kind enough to create a change record, that would be amazing.

Automatically closed - issue fixed for 2 weeks with no activity.

Hi!

When testing the patch, I've noticed that it's not always working properly.

I will try to explain. I've tested it with commerce_recurring.
The job payload is simple: ['order_id' => $order->id()]
So I've run the 'Commerce Recurring' cron job to process a recurring order. The job has created two queue items:
with types commerce_recurring_order_close and commerce_recurring_order_renew with payload {"order_id":"230185"}.
Then I've run the 'Advanced Queue' cron job. It has processed the queue items. One of them was failed due to another issue not related to this one. And another one was successfully processed. So I had two queue items with states failure and

success</code at those moment.
But after running the  'Commerce Recurring' cron job again, two new queue items with the same payload and fingerprint were created. So the items were duplicated.
And if ran the the  'Commerce Recurring' cron job again without running the the 'Advanced Queue' cron job before it, the duplicates are not created, because the items had state <code>queued

and they were found as duplicated.

So, I think we should remove condition where we check only two states in the getDuplicateJobs() method:
->condition('state', [Job::STATE_QUEUED, Job::STATE_PROCESSING, Job::STATE_FAILURE], 'IN');

Please, review and re-test if possible. I'm attaching the updated patch from #125 for 1.0 and interdiff.

Thanks in advance!

Hm... Perhaps in some cases we still want to allow queuing jobs with the same fingerprint?
One usecase I could think of is a job that notifies a 3rd party system after an entity was updated, you probably wouldn't to notify twice for the same update, but would still want to notify about subsequent updates but I'm guessing in this case the payload should contain what's supposedly unique about the update i.e. the changed timestamp for example in addition to the entity ID?

Not sure, but in the case of these recurring jobs, we never want to queue duplicate jobs...

Created new issue https://www.drupal.org/project/advancedqueue/issues/3460188 🐛 In some cases the duplicated jobs are created Active

By definition a fingerprint must be unique. Therefore it should be impossible to queue two jobs with the same fingerprint. Producers who wish to run a job after all can influence the fingerprint by including an always-unique component such as a timestamp.

@bojanz: Agreed, wanted to come back and edit my message, but Roman posted a patch in that sense, a fingerprint should be unique and it'd be the responsability of the job type to make sure it is for usecases I mentioned in #132.

Draft Change Record was created: https://www.drupal.org/node/3468315 → . Feel free to adjust and publish before making a tagged release that includes this feature. Thanks, everyone for a great job on this, hope to see it as part of the tagged release in a very short time 😉

We overlooked something: we didn't harden the logic in the queue to handle the possibility that the job plugin's handleDuplicateJobs() method might return null to cancel the job.

I opened a follow-up: 🐛 Handle nulled duplicate jobs Active