diff options
| -rw-r--r-- | deps/rabbit/src/rabbit_stream_coordinator.erl | 22 |
1 files changed, 19 insertions, 3 deletions
diff --git a/deps/rabbit/src/rabbit_stream_coordinator.erl b/deps/rabbit/src/rabbit_stream_coordinator.erl index 7e8069d866..5c23fbb51d 100644 --- a/deps/rabbit/src/rabbit_stream_coordinator.erl +++ b/deps/rabbit/src/rabbit_stream_coordinator.erl @@ -825,7 +825,10 @@ phase_update_mnesia(StreamId, Args, #{reference := QName, %% amqqueue record amqqueue:set_type_state( amqqueue:set_pid(Q, LeaderPid), Conf); - _ -> + Ts -> + S = maps:get(name, Ts, undefined), + rabbit_log:debug("~s: refusing mnesia update for stale stream id ~s, current ~s", + [?MODULE, StreamId, S]), %% if the stream id isn't a match this is a stale %% update from a previous stream incarnation for the %% same queue name and we ignore it @@ -837,10 +840,23 @@ phase_update_mnesia(StreamId, Args, #{reference := QName, rabbit_amqqueue:update(QName, Fun) end) of not_found -> + rabbit_log:debug("~s: resource for stream id ~s not found, " + "recovering from rabbit_durable_queue", + [?MODULE, StreamId]), %% This can happen during recovery + %% we need to re-initialise the queue record + %% if the stream id is a match [Q] = mnesia:dirty_read(rabbit_durable_queue, QName), - %% TODO: what is the possible return type here? - _ = rabbit_amqqueue:ensure_rabbit_queue_record_is_initialized(Fun(Q)), + case amqqueue:get_type_state(Q) of + #{name := S} when S == StreamId -> + rabbit_log:debug("~s: initializing queue record for stream id ~s", + [?MODULE, StreamId]), + _ = rabbit_amqqueue:ensure_rabbit_queue_record_is_initialized(Fun(Q)), + ok; + _ -> + ok + end, + send_self_command({mnesia_updated, StreamId, Args}); _ -> send_self_command({mnesia_updated, StreamId, Args}) |