Want to add new persistence adaptor, where to start?

[deadtrickster]

Hi,

I want to persist data in a database not supported by Ecto3. I suppose it will use polling too. So what behaviour/interface I should implement? Source or Queue?

Also, how cold starts happen (say after disaster recover - fresh nodes, database is restored from backup)? Will honeydew attempt to load all the jobs at once on each node or partition somehow, like wait cluster up (whatever it means) and a selected node will dispatch persistent jobs?

Thanks!

[koudelka]

Hey there,

The ecto queues aren't just for persistence, the actual queue mechanics are run by the database itself. The purpose of that is to keep the database as the sole authority of queue state, and hence to be able to use the distribution/replication properties of the database rather than trying to implement it in Elixir.

For example, here's where the postgres version atomically reserves a job: https://github.com/koudelka/honeydew/blob/master/lib/honeydew/sources/ecto/sql/postgres.ex#L49-L59

Which database did you want to use?

[deadtrickster]

I want to use Mongo

The purpose of that is to keep the database as the sole authority of queue state, and hence to be able to use the distribution/replication properties of the database rather than trying to implement it in Elixir.

Sounds exactly what I want.

Could you please expand on cold starts question too?

[koudelka]

Since the queue operates directly out of the database, honeydew wouldn't load jobs into a local queue, it'd keep all the state in the database itself, and rely on transactional queries to change it. So the "queue" processes themselves become essentially dumb connectors to the database. When honeydew wants to give a job to a worker, it'll ask the database to "reserve" (mark as in-progress) and return a single row.

Honeydew has two different kinds of queues:

• A fixed number of different kinds of jobs per row. You don't manually enqueue jobs for this kind of queue, the insertion of a row implies that the specified jobs need to be run. (For example, when a User is added, sending a welcome email and charging their credit card). This is the EctoQueue, it writes directly to the domain model.

• Generic queues that require jobs be enqueued by the user (the Mnesia and ErlangQueue queues are of this type).

My guess is that you're looking for the latter, but with state stored in mongo.

If that's the case, you just need to implement the PollQueue callbacks, https://github.com/koudelka/honeydew/blob/master/lib/honeydew/poll_queue.ex#L16-L22 as well as adding a new "source" to connect to mongo:

It'll probably end up looking something like this:

defmodule MongoSource do
  require Logger
  alias Honeydew.Job
  alias Honeydew.Queue

  @behaviour Queue

  @impl true
  def validate_args!([host: host, port: port, db: db]) when is_binary(host) and is_integer(port) and is_binary(db) do: :ok
  def validate_args!(_), do: raise "bad args"

  @impl true
  def init(name, [host: host, port: port, db: db]) do
    connection = Mongo.connect(host, port, db)
    collection = Mongo.collection(connection, name)
    {:ok, collection}
  end

  #
  # Enqueue/Reservee
  #

  @impl true
  def enqueue(job, collection) do
    Mongo.put(collection, to_json(job))
    collection
  end

  #
  # Reserve a job by transactionally selecting the oldest job and locking the document so others don't try to run it.
  # You'll probably want to use the current time for the lock, so you have a some way to know if a job was reserved,
  # but the reserving node completely died without being able to release the lock.
  #
  # If you can, try to let mongo handle the concept of "now", that'll reduce the chance of problems from clock desync between your nodes.
  #
  # See the following for more details: https://github.com/koudelka/honeydew/blob/master/lib/honeydew/sources/ecto_source.ex#L5
  #
  @impl true
  def reserve({pending, in_progress} = state) do
    case Mongo.get_and_update(collection, where: %{state: nil}, sort_by: :enqueued_at, set: %{reserved_at: Mongo.now(), state: "reserved"}) do
      :not_found ->
        {:empty, collection}
      job ->
        {job, collection}
    end
  end

  #
  # Ack/Nack
  #

  # Job completed successfully
  @impl true
  def ack(%Job{private: id}, collection) do
    Mongo.update(collection, id, state: "completed")
    # or, if you don't want to keep finished jobs in the database, you can just delete the document instead
    collection
  end

  # Job needs to be re-run
  @impl true
  def nack(%Job{private: id}, collection) do
    Mongo.update(collection, id, state: nil)
    collection
  end

  #
  # Helpers
  #

  @impl true
  def status(collection) do
    %{count: Mongo.count(collection),
      in_progress: Mongo.count(collection, where: %{state: "reserved"})}}
  end

  @impl true
  def filter(collection, :stale) do
    Mongo.get(collection, where: %{reserved_at: %{"$gt" => Mongo.now() + "1 hour"}})
  end

  @impl true
  def cancel(%Job{private: id}, collection) do
    reply =
      case Mongo.get_and_update(collection, where: %{id: id}, set: %{state: "cancelled"}) do
        :not_found ->
          {:error, :not_found}
        %Job{} ->
          :ok
      end

    # or, you can just remove the job from the db instead of updating its status

    {reply, collection}
  end
end

Since the possibility of "stale" jobs exists, you'll probably want to borrow the EctoSource's :__reset_stale__ functionality, to find jobs where the entire that the worker was on crashed and left the job in a "reserved" state.

Hope that helps!

[koudelka]

You might also be able to use mongo's "change streams" to listen for jobs, rather than polling the collection, that'll probably lessen the load on the db.

[deadtrickster]

Oh this definitely does help! Also startAfter from 4.2 looks so promising. I'll close this one now. Will open a new one if needed.

[scottmessinger]

@deadtrickster Did you ever get get Honeydew working with Mongo? We're looking to do the same thing and would love to use your library if it's public!