Ejabberd clustering corruption with mnesia

%%
%% Mod_vitamin: keeping your presence list healthy
%%
%% Note on use: This module is not persistent. Across restarts of the whole
%% cluster, you'll have to re-add the necessary jids. For that reason, the runtime
%% configuration is discouraged when compared to hard-coding the jids in the configuration
%% file. Any serious change to the jids made at runtime should also be made in the config file.
%%
%%
%% Authors: Dylan Ayrey and Alexander Kyte
%%

-module(mod_vitamin).

-behavior(gen_mod).

-export([start/2, stop/1, on_available/1, on_unavailable/4, runtime_config/1, query_presence_subscribers/0, remove_subscriber/1, web_menu_node/3, web_page_node/5]).
-record(presence, {key, jids=[]}).

-include("ejabberd.hrl").
-include("ejabberd_http.hrl").
-include("ejabberd_web_admin.hrl").

start(Host, _Opts) ->
  ?DEBUG("Mod_vitamin loaded", []),
  %%parse configs
  [{names, Names_to_notify}] = _Opts,
  Jids_to_notify = lists:map(fun jlib:string_to_jid/1, Names_to_notify),
  init_mnesia(Jids_to_notify), 
  %% Expose a web interface
  ejabberd_hooks:add(webadmin_menu_node, ?MODULE, web_menu_node, 50),
  ejabberd_hooks:add(webadmin_page_node, ?MODULE, web_page_node, 50),
  %% Watch and forward join/parts
  ejabberd_hooks:add(user_available_hook, Host, ?MODULE, on_available, 50),
  ejabberd_hooks:add(unset_presence_hook, Host, ?MODULE, on_unavailable, 50),
  %% Expose an ejabberdctl interface
  catch ejabberd_commands:register_commands(commands()),
  ?DEBUG("Started up okay", []),
  ok.

init_mnesia(Jids) ->
  %% For some reason this always clobbers the existing DB. The exisiting DB is never loaded, even if we
  %% include the timeout mnesia call here.
  case mnesia:create_table(presence, [{attributes, record_info(fields, presence)}, {disc_only_copies, [node()]}]) of 
      {aborted, {already_exists, _}} ->
        ok;
      {aborted, Reason} ->
        ?DEBUG("Mnesia encountered an error ~p", [Reason]);
      {atomic, ok} ->
        F = fun() ->
            ?DEBUG("First Run, trashing mnesia", []),
            mnesia:write(#presence{key="jids", jids=Jids})
        end,
        mnesia:activity(transaction, F),
        ?DEBUG("Successfully inserted. State is now ~p", query_presence_subscribers())
    end.

stop(Host) ->
  ?DEBUG("Mod_vitamin unloaded", []),
  ejabberd_hooks:delete(user_available_hook, Host, ?MODULE, on_available, 50),
  ejabberd_hooks:delete(unset_presence_hook, Host, ?MODULE, on_unavailable, 50),
  ejabberd_commands:unregister_commands(commands()),
  ejabberd_hooks:delete(webadmin_menu_node, ?MODULE, web_menu_node, 50),
  ejabberd_hooks:delete(webadmin_page_node, ?MODULE, web_page_node, 50),
  ok.

on_available(FromJid) ->
  ?DEBUG("On available hit", []),
  XmlBody = {xmlelement, "presence", [], []},
  process_presence(FromJid, XmlBody).

on_unavailable(User, Server, Resource, _) ->
  ?DEBUG("On unavailable hit", []),
  FromJid = jlib:make_jid(User, Server, Resource),
  XmlBody = {xmlelement, "presence", [{"type", "unavailable"}], []},
  process_presence(FromJid, XmlBody),
  none.

process_presence(FromJid, XmlBody) ->
  F = fun() ->
    case mnesia:read({presence, "jids"}) of
      [#presence{key="jids", jids=Jids_to_notify}] ->
        message_jids(FromJid, Jids_to_notify, XmlBody);
      [] ->
        ?DEBUG("No jids in mnesia presence table", [])
    end
  end,
  mnesia:activity(transaction, F).

message_jids(_, [], _) -> ok;
message_jids(From, [To | ToJids], XmlBody) -> 
  ?DEBUG("Sending to ~p from ~p", [jlib:jid_to_string(From), jlib:jid_to_string(To)]),
  ejabberd_router:route(From, To, XmlBody),
  message_jids(From, ToJids, XmlBody).

%% We define a runtime ejabberd command to run with ejabberdctl or mod_xmlrpc
%% ejabberd_ctl register_presence_subscriber "akyte@somehostname.net"
%%  EXAMPLE IN PYTHON
%% This is an example XML-RPC client in Python, thanks to Diddek:
%% -------
%% import xmlrpclib
%% 
%% server_url = 'http://127.0.0.1:4560';
%% server = xmlrpclib.Server(server_url);
%% 
%% params = {}
%% params["username"] = "akyte@somehostname.net"
%% result = server.register_presence_subscriber(params)
%% print result
%% -------
 
-include("ejabberd_commands.hrl").

commands() -> 
  [
    #ejabberd_commands{name = register_presence_subscriber, tags = [mod_vitamin],
      desc = "Add a user to the list of presence subscribers at runtime",
      module = ?MODULE, function = runtime_config,
      args = [{username, string}],
      result = {success, string}},
    #ejabberd_commands{name = remove_presence_subscriber, tags = [mod_vitamin],
      desc = "Add a user to the list of presence subscribers at runtime",
      module = ?MODULE, function = remove_subscriber,
      args = [{username, string}],
      result = {success, string}},
    #ejabberd_commands{name = query_presence_subscribers, tags = [mod_vitamin],
      desc = "Get the list of presence subscribers",
      module = ?MODULE, function = query_presence_subscribers,
      args = [],
      result = {names, {list, {names, string}}}} 
  ].

string_to_jid(Str, Fun) ->
  case TheJid = jlib:string_to_jid(Str) of
    error ->
      ok;
    _ ->
      Fun(TheJid)
  end,
  ok.
  

runtime_config(Username) ->
  G = fun(NewJid) ->
    F = fun() ->
      case mnesia:read({presence, "jids"}) of
        [#presence{key="jids", jids=OldJids}] ->
          %% We use usort to ensure no duplicates
          mnesia:write(#presence{key="jids", jids=lists:usort([NewJid | OldJids])}),
          "success";
        [] ->
          ?DEBUG("No jids in mnesia presence table", [])
      end
    end,
    mnesia:activity(transaction, F)
  end,
  string_to_jid(Username, G).


remove_subscriber(Username) ->
  G = fun(BadJid) ->
    ?DEBUG("Removing subscriber: ~p, with jid ~p", [Username, BadJid]),
    F = fun() ->
      case mnesia:read({presence, "jids"}) of
        [#presence{key="jids", jids=OldJids}] ->
          mnesia:write(#presence{key="jids", jids=lists:delete(BadJid, OldJids)}),
    ?DEBUG("Successfully removed ~p from ~p", [BadJid, OldJids]),
          "success";
        [] ->
          ?DEBUG("No jids in mnesia presence table", [])
      end
    end,
    mnesia:activity(transaction, F)
  end,
  string_to_jid(Username, G).

query_presence_subscribers() ->
  F = fun() ->
      case mnesia:read({presence, "jids"}) of
      [#presence{key="jids", jids=Jids}] ->
        lists:map(fun jlib:jid_to_string/1, Jids);
      [] ->
        ?DEBUG("No jids in mnesia presence table", []),
        []
    end
  end,
  mnesia:activity(transaction, F).

%% WEB LOGIC
web_menu_node(Acc, _Node, Lang) ->
  Acc ++ [{"Mod_Vitamin", ?T("Presence Subscriptions")}].

web_page_node(_, Node, ["Mod_Vitamin"], Query, Lang) ->
  Res = [?XC("h1", "Presence Subscription As a Service") | content_generation(Node, Query, Lang)],
  {stop, Res};
web_page_node(Acc, _, _, _, _) -> Acc.


content_generation(Node, Query, Lang) -> 
  process_post(Query, Node),
  Static_title =
    [?XC("p", ?T("Mod Vitamin:"))],
  Jid_to_remove_form = fun(Jid) ->
  ?XE("tr",
      [?X("td"),
         ?XCT("td", Jid),
         ?XE("td", [?INPUTS("checkbox", "delete_subscriber", Jid, "70")])
      ])
  end,
  Static_form = [?XAE("form", [{"method", "post"}], 
        [?XAE("table", [],
             [?XE("tbody",
                 lists:map(Jid_to_remove_form, query_presence_subscribers()) ++ 
                  [?XE("tr", 
                     [?X("td"),
                      ?XCT("td", "Add a subscriber"),
                      ?XE("td", [?INPUTS("text", "new_subscriber", "", "70")])
                  ]),
                  ?XE("tr", 
                     [?X("td"),
                      ?XCT("td", "Submit"),
                      ?XE("td", [?INPUTT("submit", "Submit-button", "Execute")])
                  ])
	          ]
		  )])])],
  Static_title ++ Static_form.

process_post(Query, _Node) ->
  lists:map(
	fun({"new_subscriber", ""}) -> ok;
	   ({"new_subscriber", X}) -> runtime_config(X), ok;
	   ({"delete_subscriber", X}) -> remove_subscriber(X), ok;
	   (_) -> ok
	end,
	Query),
  ok. 
  
%%
%% So we're having this bug. Whenever we need to take the cluster down to change 
%% ejabberd settings that need a restart to change, we have a ~50% chance of the
%% presences table getting corrupt. Namely, on start we see this error message:
%%
%% =ERROR REPORT==== 10-Feb-2014::14:49:21 ===
Mnesia('ejabberd@*this node's hostname*'): ** ERROR ** (core dumped to file: "/opt/ejabberd-2.1.5/bin/MnesiaCore.ejabberd@*this node's hostname*_1392_43761_551172")
 ** FATAL ** Failed to merge schema: Bad cookie in table definition presence: 'ejabberd@*this node's hostname*' = {cstruct,presence,set,[],[],['ejabberd@*other node's hostname*'],0,read_write,[],[],false,presence,[key,jids],[],[],{{1391,807862,776322},'ejabberd@*other node's hostname*'},{{2,0},[]}}, 'ejabberd@*other/master node's hostname*' = {cstruct,presence,set,[],[],['ejabberd@*other/master node's hostname*'],0,read_write,[],[],false,presence,[key,jids],[],[],{{1391,807891,609438},'ejabberd@*other/master node's hostname*'},{{2,0},[]}}
%%
%%
%% Most of the time we can do the temporary fix if it only happens on one node,
%% but we've had to completely rebuild our testing server once already. Simply 
%% removing the corrupted files constantly isn't tenable. It only seems to be happening
%% to our table, so it sounds like an issue with our use. That said, I've seen a similar
%% error message all over the rabbitmq help lists and the response is always the temporary fix
%% of removing and remaking the tables. What are we doing wrong?