not_found error on attempt to declare a queue that exists on a stopped node

this prevents loss or duplication of messages when that node recovers

I considered returing {ok, Q} | {error, not_found} from
internal_declare (and subsequently declare), and let the channel deal
with throwing the appropriate amqp error. But
- that would be a change to a crucial internal API that is used by
quite a few extensions
- it would make the return type inconsistent with
rabbit_exchange:declare, which too only returns X, not {ok, X}.
- the rabbit_amqqueue module already knows about amqp errors - see
with_or_die. So this is not breaking any concern separation.

Author: Matthias Radestock

src/rabbit_amqqueue.erl

@@ -163,15 +163,23 @@ internal_declare(Q = #amqqueue{name = QueueName}, WantDefaultBinding) ->
     case rabbit_misc:execute_mnesia_transaction(
            fun () ->
                    case mnesia:wread({rabbit_queue, QueueName}) of
-                       [] -> ok = store_queue(Q),
-                             case WantDefaultBinding of
-                                 true -> add_default_binding(Q);
-                                 false -> ok
-                             end,
-                             Q;
-                       [ExistingQ] -> ExistingQ
+                       [] ->
+                           case mnesia:read(
+                                  {rabbit_durable_queue, QueueName}) of
+                               []  -> ok = store_queue(Q),
+                                      case WantDefaultBinding of
+                                          true  -> add_default_binding(Q);
+                                          false -> ok
+                                      end,
+                                      Q;
+                               [_] -> not_found %% existing Q on stopped node
+                           end;
+                       [ExistingQ] ->
+                           ExistingQ
                    end
            end) of
+        not_found -> exit(Q#amqqueue.pid, shutdown),
+                     rabbit_misc:not_found(QueueName);
         Q         -> Q;
         ExistingQ -> exit(Q#amqqueue.pid, shutdown),
                      ExistingQ