Reevaluate reified policies on upgrade.

3 files changed, 47 insertions, 3 deletions

diff --git a/src/rabbit.erl b/src/rabbit.erl
index dddf6f47..569c5385 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -581,6 +581,7 @@ boot_delegate() ->
     rabbit_sup:start_supervisor_child(delegate_sup, [Count]).
 
 recover() ->
+    rabbit_policy:recover(),
     Qs = rabbit_amqqueue:recover(),
     ok = rabbit_binding:recover(rabbit_exchange:recover(),
                                 [QName || #amqqueue{name = QName} <- Qs]),
diff --git a/src/rabbit_policy.erl b/src/rabbit_policy.erl
index 4c5323f9..bafb017b 100644
--- a/src/rabbit_policy.erl
+++ b/src/rabbit_policy.erl
@@ -25,6 +25,7 @@
 -import(rabbit_misc, [pget/2]).
 
 -export([register/0]).
+-export([invalidate/0, recover/0]).
 -export([name/1, get/2, set/1]).
 -export([validate/4, notify/4, notify_clear/3]).
 -export([parse_set/5, set/6, delete/2, lookup/2, list/0, list/1,
@@ -51,6 +52,10 @@ set(X = #exchange{name = Name}) -> rabbit_exchange_decorator:set(
 
 set0(Name = #resource{virtual_host = VHost}) -> match(Name, list(VHost)).
 
+set(Q = #amqqueue{name = Name}, Ps) -> Q#amqqueue{policy = match(Name, Ps)};
+set(X = #exchange{name = Name}, Ps) -> rabbit_exchange_decorator:set(
+                                         X#exchange{policy = match(Name, Ps)}).
+
 get(Name, #amqqueue{policy = Policy}) -> get0(Name, Policy);
 get(Name, #exchange{policy = Policy}) -> get0(Name, Policy);
 %% Caution - SLOW.
@@ -68,6 +73,41 @@ get0(Name, List)       -> case pget(definition, List) of
 
 %%----------------------------------------------------------------------------
 
+%% Gets called during upgrades - therefore must not assume anything about the
+%% state of Mnesia
+invalidate() ->
+    rabbit_file:write_file(invalid_file(), <<"">>).
+
+recover() ->
+    case rabbit_file:is_file(invalid_file()) of
+        true  -> recover0(),
+                 rabbit_file:delete(invalid_file());
+        false -> ok
+    end.
+
+%% To get here we have to have just completed an Mnesia upgrade - i.e. we are
+%% the first node starting. So we can rewrite the whole database.  Note that
+%% recovery has not yet happened; we must work with the rabbit_durable_<thing>
+%% variants.
+recover0() ->
+    Xs = mnesia:dirty_match_object(rabbit_durable_exchange, #exchange{_ = '_'}),
+    Qs = mnesia:dirty_match_object(rabbit_durable_queue,    #amqqueue{_ = '_'}),
+    Policies = list(),
+    [rabbit_misc:execute_mnesia_transaction(
+       fun () ->
+               mnesia:write(rabbit_durable_exchange, set(X, Policies), write)
+       end) || X <- Xs],
+    [rabbit_misc:execute_mnesia_transaction(
+       fun () ->
+               mnesia:write(rabbit_durable_queue, set(Q, Policies), write)
+       end) || Q <- Qs],
+    ok.
+
+invalid_file() ->
+    filename:join(rabbit_mnesia:dir(), "policies_are_invalid").
+
+%%----------------------------------------------------------------------------
+
 parse_set(VHost, Name, Pattern, Definition, undefined) ->
     parse_set0(VHost, Name, Pattern, Definition, 0);
 parse_set(VHost, Name, Pattern, Definition, Priority) ->
@@ -211,9 +251,10 @@ match(Name, Policies) ->
         [Policy | _Rest] -> Policy
     end.
 
-matches(#resource{name = Name, kind = Kind}, Policy) ->
+matches(#resource{name = Name, kind = Kind, virtual_host = VHost}, Policy) ->
     matches_type(Kind, pget('apply-to', Policy)) andalso
-        match =:= re:run(Name, pget(pattern, Policy), [{capture, none}]).
+        match =:= re:run(Name, pget(pattern, Policy), [{capture, none}]) andalso
+        VHost =:= pget(vhost, Policy).
 
 matches_type(exchange, <<"exchanges">>) -> true;
 matches_type(queue,    <<"queues">>)    -> true;
diff --git a/src/rabbit_upgrade_functions.erl b/src/rabbit_upgrade_functions.erl
index 76bfe28a..b9ba4d76 100644
--- a/src/rabbit_upgrade_functions.erl
+++ b/src/rabbit_upgrade_functions.erl
@@ -310,7 +310,9 @@ policy_apply_to() ->
           ({runtime_parameters, Key, Value}) ->
               {runtime_parameters, Key, Value}
       end,
-      [key, value]).
+      [key, value]),
+    rabbit_policy:invalidate(),
+    ok.
 
 apply_to(Def) ->
     case [proplists:get_value(K, Def) ||