issue #49, #55, #58: ensure mnesia cache persistence over node failures

The deduplication cache was created on the node where the
resource (exchange or queue) was initialized.

If the node crashed and was replaced mnesia could not recover or
recreate the table used as a cache.

After several attempts, it seems the best way to prevent this from
happening is to replicate the table over more nodes to ensure mnesia
can recover it in case of a node disappearing.

To ensure safe enough persistence without replicating too
aggressively, the cache is created on 2/3 of the total cluster nodes.

Signed-off-by: Matteo Cafasso <[email protected]>

Author: noxdafox

lib/cache.ex

@@ -140,35 +140,23 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Cache do
 
   ## Utility functions
 
-  # Run Mnesia creation functions handling output
-  defmacro mnesia_create(function) do
-    quote do
-      case unquote(function) do
-        {:atomic, :ok} -> :ok
-        {:aborted, {:already_exists, _}} -> :ok
-        {:aborted, {:already_exists, _, _}} -> :ok
-        error -> error
-      end
-    end
-  end
-
   # Mnesia cache table creation.
   defp cache_create(cache, options) do
     persistence = case Keyword.get(options, :persistence) do
                     :disk -> :disc_copies
                     :memory -> :ram_copies
                   end
     options = [{:attributes, [:entry, :expiration]},
-               {persistence, [node()]},
+               {persistence, cache_replicas()},
                {:index, [:expiration]},
                {:user_properties, [{:limit, Keyword.get(options, :size)},
                                    {:default_ttl, Keyword.get(options, :ttl)}]}]
 
-    with :ok <- mnesia_create(Mnesia.create_table(cache, options)),
-         :ok <- mnesia_create(Mnesia.add_table_copy(cache, node(), persistence))
-    do
-      Mnesia.wait_for_tables([cache], @table_wait_time)
-    else
+    case Mnesia.create_table(cache, options) do
+      {:atomic, :ok} ->
+        Mnesia.wait_for_tables([cache], @table_wait_time)
+      {:aborted, {:already_exists, _}} ->
+        Mnesia.wait_for_tables([cache], @table_wait_time)
       error -> error
     end
   end
@@ -219,4 +207,11 @@ defmodule RabbitMQ.MessageDeduplicationPlugin.Cache do
 
     entry
   end
+
+  # List the nodes on which to create the cache replicas.
+  # Cache is replicated on two-third of the cluster nodes.
+  defp cache_replicas() do
+    nodes = [Node.self() | Node.list()]
+    nodes |> Enum.split(round((length(nodes) * 2) / 3)) |> elem(0)
+  end
 end