Merge bug24728

10 files changed, 102 insertions, 73 deletions

diff --git a/packaging/RPMS/Fedora/Makefile b/packaging/RPMS/Fedora/Makefile
index f1925757..234fc2c7 100644
--- a/packaging/RPMS/Fedora/Makefile
+++ b/packaging/RPMS/Fedora/Makefile
@@ -15,9 +15,11 @@ endif
 ifeq "$(RPM_OS)" "suse"
 REQUIRES=/sbin/chkconfig /sbin/service
 OS_DEFINES=--define '_initrddir /etc/init.d' --define 'dist .suse'
+START_PROG=setsid
 else
 REQUIRES=chkconfig initscripts
 OS_DEFINES=--define '_initrddir /etc/rc.d/init.d'
+START_PROG=runuser rabbitmq --session-command
 endif
 
 rpms:   clean server
@@ -32,7 +34,7 @@ prepare:
 	cp ${COMMON_DIR}/* SOURCES/
 	sed -i \
 	    -e 's|^LOCK_FILE=.*$$|LOCK_FILE=/var/lock/subsys/$$NAME|' \
-	    -e 's|^START_PROG=.*$$|START_PROG="runuser rabbitmq --session-command"|' \
+	    -e 's|^START_PROG=.*$$|START_PROG="$(START_PROG)"|' \
 	    SOURCES/rabbitmq-server.init
 ifeq "$(RPM_OS)" "fedora"
 # Fedora says that only vital services should have Default-Start
diff --git a/packaging/windows/Makefile b/packaging/windows/Makefile
index a910941b..1c222162 100644
--- a/packaging/windows/Makefile
+++ b/packaging/windows/Makefile
@@ -8,10 +8,7 @@ dist:
 	$(MAKE) -C $(SOURCE_DIR)
 
 	mkdir $(SOURCE_DIR)/sbin
-	mv $(SOURCE_DIR)/scripts/rabbitmq-server.bat $(SOURCE_DIR)/sbin
-	mv $(SOURCE_DIR)/scripts/rabbitmq-service.bat $(SOURCE_DIR)/sbin
-	mv $(SOURCE_DIR)/scripts/rabbitmq-plugins.bat $(SOURCE_DIR)/sbin
-	mv $(SOURCE_DIR)/scripts/rabbitmqctl.bat $(SOURCE_DIR)/sbin
+	mv $(SOURCE_DIR)/scripts/*.bat $(SOURCE_DIR)/sbin
 	rm -rf $(SOURCE_DIR)/scripts
 	rm -rf $(SOURCE_DIR)/codegen* $(SOURCE_DIR)/Makefile $(SOURCE_DIR)/*mk
 	rm -f $(SOURCE_DIR)/README
diff --git a/src/credit_flow.erl b/src/credit_flow.erl
index 072f4d9d..ba99811f 100644
--- a/src/credit_flow.erl
+++ b/src/credit_flow.erl
@@ -96,7 +96,8 @@ peer_down(Peer) ->
     %% credit_deferred and thus send messages into the void...
     unblock(Peer),
     erase({credit_from, Peer}),
-    erase({credit_to, Peer}).
+    erase({credit_to, Peer}),
+    ok.
 
 %% --------------------------------------------------------------------------
 
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 106a9960..83d9ae22 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -24,9 +24,6 @@
 -define(SYNC_INTERVAL,                 25). %% milliseconds
 -define(RAM_DURATION_UPDATE_INTERVAL,  5000).
 
--define(BASE_MESSAGE_PROPERTIES,
-        #message_properties{expiry = undefined, needs_confirming = false}).
-
 -export([start_link/1, info_keys/0]).
 
 -export([init_with_backing_queue_state/7]).
@@ -333,12 +330,10 @@ ensure_expiry_timer(State = #q{expires = undefined}) ->
     State;
 ensure_expiry_timer(State = #q{expires = Expires}) ->
     case is_unused(State) of
-        true ->
-            NewState = stop_expiry_timer(State),
-            TRef = erlang:send_after(Expires, self(), maybe_expire),
-            NewState#q{expiry_timer_ref = TRef};
-        false ->
-            State
+        true  -> NewState = stop_expiry_timer(State),
+                 TRef = erlang:send_after(Expires, self(), maybe_expire),
+                 NewState#q{expiry_timer_ref = TRef};
+        false -> State
     end.
 
 ensure_stats_timer(State) ->
@@ -532,15 +527,10 @@ attempt_delivery(Delivery = #delivery{sender     = SenderPid,
         {false, BQS1} ->
             DeliverFun =
                 fun (AckRequired, State1 = #q{backing_queue_state = BQS2}) ->
-                        %% we don't need an expiry here because
-                        %% messages are not being enqueued, so we use
-                        %% an empty message_properties.
-                        {AckTag, BQS3} =
-                            BQ:publish_delivered(
-                              AckRequired, Message,
-                              (?BASE_MESSAGE_PROPERTIES)#message_properties{
-                                needs_confirming = needs_confirming(Confirm)},
-                              SenderPid, BQS2),
+                        Props = message_properties(Confirm, State1),
+                        {AckTag, BQS3} = BQ:publish_delivered(
+                                           AckRequired, Message, Props,
+                                           SenderPid, BQS2),
                         {{Message, false, AckTag}, true,
                          State1#q{backing_queue_state = BQS3}}
                 end,
@@ -567,8 +557,7 @@ deliver_or_enqueue(Delivery = #delivery{message = Message,
         maybe_record_confirm_message(Confirm, State1),
     case Delivered of
         true  -> State2;
-        false -> Props = (message_properties(State)) #message_properties{
-                           needs_confirming = needs_confirming(Confirm)},
+        false -> Props = message_properties(Confirm, State),
                  BQS1 = BQ:publish(Message, Props, SenderPid, BQS),
                  ensure_ttl_timer(State2#q{backing_queue_state = BQS1})
     end.
@@ -696,8 +685,9 @@ discard_delivery(#delivery{sender = SenderPid,
                             backing_queue_state = BQS}) ->
     State#q{backing_queue_state = BQ:discard(Message, SenderPid, BQS)}.
 
-message_properties(#q{ttl=TTL}) ->
-    #message_properties{expiry = calculate_msg_expiry(TTL)}.
+message_properties(Confirm, #q{ttl = TTL}) ->
+    #message_properties{expiry           = calculate_msg_expiry(TTL),
+                        needs_confirming = needs_confirming(Confirm)}.
 
 calculate_msg_expiry(undefined) -> undefined;
 calculate_msg_expiry(TTL)       -> now_micros() + (TTL * 1000).
@@ -735,10 +725,9 @@ dead_letter_fun(Reason, _State) ->
 
 dead_letter_msg(Msg, AckTag, Reason, State = #q{dlx = DLX}) ->
     case rabbit_exchange:lookup(DLX) of
-        {error, not_found} ->
-            noreply(State);
-        _ ->
-            dead_letter_msg_existing_dlx(Msg, AckTag, Reason, State)
+        {error, not_found} -> noreply(State);
+        _                  -> dead_letter_msg_existing_dlx(Msg, AckTag, Reason,
+                                                           State)
     end.
 
 dead_letter_msg_existing_dlx(Msg, AckTag, Reason,
@@ -755,7 +744,7 @@ dead_letter_msg_existing_dlx(Msg, AckTag, Reason,
     State1 = lists:foldl(fun monitor_queue/2, State, QPids),
     State2 = State1#q{publish_seqno = MsgSeqNo + 1},
     case QPids of
-        [] -> {_, BQS1} = BQ:ack([AckTag], BQS),
+        [] -> {_Guids, BQS1} = BQ:ack([AckTag], BQS),
               cleanup_after_confirm(State2#q{backing_queue_state = BQS1});
         _  -> State3 =
                   lists:foldl(
@@ -792,11 +781,12 @@ handle_queue_down(QPid, Reason, State = #q{queue_monitors = QMons,
         error ->
             noreply(State);
         {ok, _} ->
-            #resource{name = QName} = qname(State),
-            rabbit_log:info("DLQ ~p (for ~p) died~n", [QPid, QName]),
+            rabbit_log:info("DLQ ~p (for ~s) died~n",
+                            [QPid, rabbit_misc:rs(qname(State))]),
+            State1 = State#q{queue_monitors = dict:erase(QPid, QMons)},
             case gb_trees:lookup(QPid, UQM) of
                 none ->
-                    noreply(State);
+                    noreply(State1);
                 {value, MsgSeqNosSet} ->
                     case rabbit_misc:is_abnormal_termination(Reason) of
                         true  -> rabbit_log:warning(
@@ -804,9 +794,7 @@ handle_queue_down(QPid, Reason, State = #q{queue_monitors = QMons,
                                    [gb_sets:size(MsgSeqNosSet)]);
                         false -> ok
                     end,
-                    handle_confirm(gb_sets:to_list(MsgSeqNosSet), QPid,
-                                   State#q{queue_monitors =
-                                               dict:erase(QPid, QMons)})
+                    handle_confirm(gb_sets:to_list(MsgSeqNosSet), QPid, State1)
             end
     end.
 
@@ -1285,14 +1273,14 @@ handle_cast({ack, AckTags, ChPid}, State) ->
 handle_cast({reject, AckTags, Requeue, ChPid}, State) ->
     noreply(subtract_acks(
               ChPid, AckTags, State,
-              fun (State1 = #q{backing_queue       = BQ,
-                               backing_queue_state = BQS}) ->
-                      case Requeue of
-                          true  -> requeue_and_run(AckTags, State1);
-                          false -> Fun = dead_letter_fun(rejected, State),
+              case Requeue of
+                  true  -> fun (State1) -> requeue_and_run(AckTags, State1) end;
+                  false -> Fun = dead_letter_fun(rejected, State),
+                           fun (State1 = #q{backing_queue       = BQ,
+                                            backing_queue_state = BQS}) ->
                                    BQS1 = BQ:fold(Fun, BQS, AckTags),
                                    State1#q{backing_queue_state = BQS1}
-                      end
+                           end
               end));
 
 handle_cast(delete_immediately, State) ->
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index bb6636eb..cac622f8 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -1411,11 +1411,10 @@ process_routing_result(routed,    QPids, XName,  MsgSeqNo,   _, State) ->
 send_nacks([], State) ->
     State;
 send_nacks(MXs, State = #ch{tx_status = none}) ->
-    MsgSeqNos = [ MsgSeqNo || {MsgSeqNo, _} <- MXs ],
-    coalesce_and_send(MsgSeqNos,
+    coalesce_and_send([MsgSeqNo || {MsgSeqNo, _} <- MXs],
                       fun(MsgSeqNo, Multiple) ->
                               #'basic.nack'{delivery_tag = MsgSeqNo,
-                                            multiple = Multiple}
+                                            multiple     = Multiple}
                       end, State);
 send_nacks(_, State) ->
     maybe_complete_tx(State#ch{tx_status = failed}).
diff --git a/src/rabbit_guid.erl b/src/rabbit_guid.erl
index f4c425ca..ba0cb04f 100644
--- a/src/rabbit_guid.erl
+++ b/src/rabbit_guid.erl
@@ -19,6 +19,7 @@
 -behaviour(gen_server).
 
 -export([start_link/0]).
+-export([filename/0]).
 -export([gen/0, gen_secure/0, string/2, binary/2]).
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
@@ -38,6 +39,7 @@
 -type(guid() :: binary()).
 
 -spec(start_link/0 :: () -> rabbit_types:ok_pid_or_error()).
+-spec(filename/0 :: () -> string()).
 -spec(gen/0 :: () -> guid()).
 -spec(gen_secure/0 :: () -> guid()).
 -spec(string/2 :: (guid(), any()) -> string()).
@@ -51,8 +53,14 @@ start_link() ->
     gen_server:start_link({local, ?SERVER}, ?MODULE,
                           [update_disk_serial()], []).
 
+%% We use this to detect a (possibly rather old) Mnesia directory,
+%% since it has existed since at least 1.7.0 (as far back as I cared
+%% to go).
+filename() ->
+    filename:join(rabbit_mnesia:dir(), ?SERIAL_FILENAME).
+
 update_disk_serial() ->
-    Filename = filename:join(rabbit_mnesia:dir(), ?SERIAL_FILENAME),
+    Filename = filename(),
     Serial = case rabbit_file:read_term_file(Filename) of
                  {ok, [Num]}     -> Num;
                  {error, enoent} -> 0;
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 4d419fd9..c714d3a7 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -508,7 +508,7 @@ init_db(ClusterNodes, Force) ->
                   ok                    -> ok;
                   %% If we're just starting up a new node we won't have a
                   %% version
-                  version_not_available -> ok = rabbit_version:record_desired()
+                  starting_from_scratch -> ok = rabbit_version:record_desired()
               end
       end).
 
diff --git a/src/rabbit_upgrade.erl b/src/rabbit_upgrade.erl
index 80f50b38..e1a7bcae 100644
--- a/src/rabbit_upgrade.erl
+++ b/src/rabbit_upgrade.erl
@@ -28,7 +28,9 @@
 -ifdef(use_specs).
 
 -spec(maybe_upgrade_mnesia/0 :: () -> 'ok').
--spec(maybe_upgrade_local/0 :: () -> 'ok' | 'version_not_available').
+-spec(maybe_upgrade_local/0 :: () -> 'ok' |
+                                     'version_not_available' |
+                                     'starting_from_scratch').
 
 -endif.
 
@@ -119,8 +121,13 @@ remove_backup() ->
     info("upgrades: Mnesia backup removed~n", []).
 
 maybe_upgrade_mnesia() ->
-    AllNodes = rabbit_mnesia:all_clustered_nodes(),
+    %% rabbit_mnesia:all_clustered_nodes/0 will return [] at this point
+    %% if we are a RAM node since Mnesia has not started yet.
+    AllNodes = lists:usort(rabbit_mnesia:all_clustered_nodes() ++
+                               rabbit_mnesia:read_cluster_nodes_config()),
     case rabbit_version:upgrades_required(mnesia) of
+        {error, starting_from_scratch} ->
+            ok;
         {error, version_not_available} ->
             case AllNodes of
                 [_] -> ok;
@@ -235,6 +242,7 @@ nodes_running(Nodes) ->
 maybe_upgrade_local() ->
     case rabbit_version:upgrades_required(local) of
         {error, version_not_available} -> version_not_available;
+        {error, starting_from_scratch} -> starting_from_scratch;
         {error, _} = Err               -> throw(Err);
         {ok, []}                       -> ensure_backup_removed(),
                                           ok;
diff --git a/src/rabbit_version.erl b/src/rabbit_version.erl
index 7545d813..1cc7d6c8 100644
--- a/src/rabbit_version.erl
+++ b/src/rabbit_version.erl
@@ -96,7 +96,10 @@ record_desired_for_scope(Scope) ->
 upgrades_required(Scope) ->
     case recorded_for_scope(Scope) of
         {error, enoent} ->
-            {error, version_not_available};
+            case filelib:is_file(rabbit_guid:filename()) of
+                false -> {error, starting_from_scratch};
+                true  -> {error, version_not_available}
+            end;
         {ok, CurrentHeads} ->
             with_upgrade_graph(
               fun (G) ->
diff --git a/src/supervisor2.erl b/src/supervisor2.erl
index a2f4fae9..8dd8aba8 100644
--- a/src/supervisor2.erl
+++ b/src/supervisor2.erl
@@ -9,15 +9,15 @@
 %%    terminated as per the shutdown component of the child_spec.
 %%
 %% 3) child specifications can contain, as the restart type, a tuple
-%%    {permanent, Delay} | {transient, Delay} where Delay >= 0. The
-%%    delay, in seconds, indicates what should happen if a child, upon
-%%    being restarted, exceeds the MaxT and MaxR parameters. Thus, if
-%%    a child exits, it is restarted as normal. If it exits
-%%    sufficiently quickly and often to exceed the boundaries set by
-%%    the MaxT and MaxR parameters, and a Delay is specified, then
-%%    rather than stopping the supervisor, the supervisor instead
-%%    continues and tries to start up the child again, Delay seconds
-%%    later.
+%%    {permanent, Delay} | {transient, Delay} | {intrinsic, Delay}
+%%    where Delay >= 0 (see point (4) below for intrinsic). The delay,
+%%    in seconds, indicates what should happen if a child, upon being
+%%    restarted, exceeds the MaxT and MaxR parameters. Thus, if a
+%%    child exits, it is restarted as normal. If it exits sufficiently
+%%    quickly and often to exceed the boundaries set by the MaxT and
+%%    MaxR parameters, and a Delay is specified, then rather than
+%%    stopping the supervisor, the supervisor instead continues and
+%%    tries to start up the child again, Delay seconds later.
 %%
 %%    Note that you can never restart more frequently than the MaxT
 %%    and MaxR parameters allow: i.e. you must wait until *both* the
@@ -31,6 +31,16 @@
 %%    the MaxT and MaxR parameters to permit the child to be
 %%    restarted. This may require waiting for longer than Delay.
 %%
+%%    Sometimes, you may wish for a transient or intrinsic child to
+%%    exit abnormally so that it gets restarted, but still log
+%%    nothing. gen_server will log any exit reason other than
+%%    'normal', 'shutdown' or {'shutdown', _}. Thus the exit reason of
+%%    {'shutdown', 'restart'} is interpreted to mean you wish the
+%%    child to be restarted according to the delay parameters, but
+%%    gen_server will not log the error. Thus from gen_server's
+%%    perspective it's a normal exit, whilst from supervisor's
+%%    perspective, it's an abnormal exit.
+%%
 %% 4) Added an 'intrinsic' restart type. Like the transient type, this
 %%    type means the child should only be restarted if the child exits
 %%    abnormally. Unlike the transient type, if the child exits
@@ -529,25 +539,23 @@ restart_child(Pid, Reason, State) ->
 	    {ok, State}
     end.
 
-do_restart({RestartType, Delay}, Reason, Child, State) ->
-    case restart1(Child, State) of
-        {ok, NState} ->
-            {ok, NState};
-        {terminate, NState} ->
-            _TRef = erlang:send_after(trunc(Delay*1000), self(),
-				      {delayed_restart,
-				       {{RestartType, Delay}, Reason, Child}}),
-            {ok, state_del_child(Child, NState)}
-    end;
+do_restart({permanent = RestartType, Delay}, Reason, Child, State) ->
+    do_restart_delay({RestartType, Delay}, Reason, Child, State);
 do_restart(permanent, Reason, Child, State) ->
     report_error(child_terminated, Reason, Child, State#state.name),
     restart(Child, State);
 do_restart(Type, normal, Child, State) ->
     del_child_and_maybe_shutdown(Type, Child, State);
+do_restart({RestartType, Delay}, {shutdown, restart} = Reason, Child, State)
+  when RestartType =:= transient orelse RestartType =:= intrinsic ->
+    do_restart_delay({RestartType, Delay}, Reason, Child, State);
 do_restart(Type, {shutdown, _}, Child, State) ->
     del_child_and_maybe_shutdown(Type, Child, State);
 do_restart(Type, shutdown, Child = #child{child_type = supervisor}, State) ->
     del_child_and_maybe_shutdown(Type, Child, State);
+do_restart({RestartType, Delay}, Reason, Child, State)
+  when RestartType =:= transient orelse RestartType =:= intrinsic ->
+    do_restart_delay({RestartType, Delay}, Reason, Child, State);
 do_restart(Type, Reason, Child, State) when Type =:= transient orelse
                                             Type =:= intrinsic ->
     report_error(child_terminated, Reason, Child, State#state.name),
@@ -557,8 +565,21 @@ do_restart(temporary, Reason, Child, State) ->
     NState = state_del_child(Child, State),
     {ok, NState}.
 
+do_restart_delay({RestartType, Delay}, Reason, Child, State) ->
+    case restart1(Child, State) of
+        {ok, NState} ->
+            {ok, NState};
+        {terminate, NState} ->
+            _TRef = erlang:send_after(trunc(Delay*1000), self(),
+                                      {delayed_restart,
+                                       {{RestartType, Delay}, Reason, Child}}),
+            {ok, state_del_child(Child, NState)}
+    end.
+
 del_child_and_maybe_shutdown(intrinsic, Child, State) ->
     {shutdown, state_del_child(Child, State)};
+del_child_and_maybe_shutdown({intrinsic, _Delay}, Child, State) ->
+    {shutdown, state_del_child(Child, State)};
 del_child_and_maybe_shutdown(_, Child, State) ->
     {ok, state_del_child(Child, State)}.
 
@@ -911,7 +932,8 @@ supname(N,_)      -> N.
 %%%       Func is {Mod, Fun, Args} == {atom, atom, list}
 %%%       RestartType is permanent | temporary | transient |
 %%%                      intrinsic | {permanent, Delay} |
-%%%                      {transient, Delay} where Delay >= 0
+%%%                      {transient, Delay} | {intrinsic, Delay}
+%%                       where Delay >= 0
 %%%       Shutdown = integer() | infinity | brutal_kill
 %%%       ChildType = supervisor | worker
 %%%       Modules = [atom()] | dynamic
@@ -962,6 +984,7 @@ validRestartType(temporary)          -> true;
 validRestartType(transient)          -> true;
 validRestartType(intrinsic)          -> true;
 validRestartType({permanent, Delay}) -> validDelay(Delay);
+validRestartType({intrinsic, Delay}) -> validDelay(Delay);
 validRestartType({transient, Delay}) -> validDelay(Delay);
 validRestartType(RestartType)        -> throw({invalid_restart_type,
                                                RestartType}).