Redesigning API for concurrency

[lguminski]

For my use case, before placing a new job in the queue, I want to cancel all jobs in the queue with the same arguments.

I can see that Honeydew example suggests the following

honeydew/examples/filter_and_cancel.exs

Lines 37 to 39 in 03a8d45

	Honeydew.filter(:my_queue, %{task: {:run, [10]}})
	|> List.first
	|> Honeydew.cancel

I believe that there is a flaw in this code, because filter and cancel are two consecutive commands and after filter returned data and before cancel gets executed there are many things that can happen in a concurrent system. Jobs can get cancelled, start getting processed, complete, or new jobs can start that match the search criterion. Shortly speaking cancel is likely to operate based on outdated input.

One of possible solutions to this problem is something similar to Map.get_and_update/3. What I mean is that the filter function could be extended with one more parameter - a function executed in its context

filter(queue_name(), filter(), function()) :: :ok | {:error, :failed}

 Honeydew.filter(:my_queue, %{task: {:run, [10]}}, fn job -> Honeydew.cancel(job) end)

or even better adding also an accumulator as a parameter (similar to Enum.reduce/3)

filter(queue_name(), filter(), acc(), function()) :: acc()

the accumulator could be used to keep track on cancellations, which is needed for scenario described in examples/filter_and_cancel.exs

 was_executed = Honeydew.filter(:my_queue, %{task: {:run, [10]}}, false, fn job, was_executed ->
        if not was_executed do
          Honeydew.cancel(job)
        end
        true
      end)

Another option are transactions.