AMQP 1.0 client: gracefully handle connection termination in more places (#17013) (#17015)

* amqp10_client: handle DOWN in more states

When running AMQP 1.0 shovels, sometimes they would crash with errors
like this:

```
  reason: {function_clause,
              [{amqp10_client_connection,hdr_sent,
                   [info,
                    {'DOWN',#Ref<0.2037148225.2260205569.233493>,process,
                        <0.29675.0>,normal},
                    {state,0,<0.29672.0>,
                        #Ref<0.2037148225.2260205569.233493>,<0.29676.0>,
                        [{<0.29530.0>,
                          [alias|
                           #Ref<0.0.3779843.2037148225.2260271105.233488>]}],
                        <0.29675.0>,
                        {tcp,#Port<0.396>},
                        undefined,undefined,
                        #{notify => <0.29530.0>,port => 5672,
                          address => "green",
                          sasl =>
                              {encrypted,
                                  <<"6APRufcR3aNjXGKFDynwAyf37GF3YqrJq06ecxGXPXNcdqSEXEKXoitUXhMTLBRhzdnWAQmdu9lw2aZ5TlxChT04sNM7XiCPZVhVFwKI124OLrxsEKjS0zOvYZc8wAlT02XqNHL4HGv9YHRGNX2aFpYCsLVTdEL4yRA8COvRomXsCDuCJ1XpWNvo/Y6fTjkW9PzfZiOwPWyfyvBx2nVCMw==">>},
                          hostname => <<"green">>,
                          properties =>
                              #{<<"ignore-maintenance">> => {boolean,true}},
                          notify_when_opened => <0.29530.0>,
                          notify_when_closed => <0.29530.0>,
                          transfer_limit_margin => 0,
                          max_frame_size => 1048576}}],
                   [{file,"amqp10_client_connection.erl"},{line,244}]},
               {gen_statem,loop_state_callback,11,
                   [{file,"gen_statem.erl"},{line,3735}]},
               {proc_lib,init_p_do_apply,3,
                   [{file,"proc_lib.erl"},{line,329}]}]}
```

* amqp10_client: handle unexpected EXIT reasons in close_sent

Connections linked by their owner (e.g. AMQP 1.0 shovels call
link/1 on the connection PID) could crash with a function_clause
error if that owner exited with a reason other than shutdown while
the connection was in the close_sent state:

```
  reason: {function_clause,
               [{amqp10_client_connection,close_sent,
                    [info,
                     {'EXIT',<0.11280.0>,killed},
                     {state,1,<0.11283.0>,
                         #Ref<0.1827165118.3650617346.181159>,<0.11287.0>,[],
                         <0.11285.0>,
                         {tcp,#Port<0.308>},
                         15000,
                         {once,#Ref<0.1827165118.3651141633.134174>},
                         #{notify => <0.11280.0>,port => 5672,
                           address => "green",
                           hostname => <<"green">>,
                           notify_when_opened => <0.11280.0>,
                           notify_when_closed => <0.11280.0>}}],
                    [{file,"amqp10_client_connection.erl"},{line,350}]},
                {gen_statem,loop_state_callback,11,
                    [{file,"gen_statem.erl"},{line,3735}]},
                {proc_lib,init_p_do_apply,3,
                    [{file,"proc_lib.erl"},{line,329}]}]}
```

close_sent/3 only matched EXIT signals with reason shutdown or
{shutdown, _}, which is what our own supervisor sends when it wants
us to terminate gracefully within its own timeout. Any other EXIT
reason, such as an owning process being killed, now stops the
connection instead, mirroring how we already give up waiting for a
close frame when the reader process goes down.

(cherry picked from commit f815505)

Co-authored-by: Michal Kuratczyk <michal.kuratczyk@broadcom.com>

Author: mergify[bot]

deps/amqp10_client/src/amqp10_client_connection.erl

@@ -227,7 +227,10 @@ sasl_hdr_rcvds(_EvtType, #'v1_0.sasl_mechanisms'{
 sasl_hdr_rcvds({call, From}, begin_session,
                #state{pending_session_reqs = PendingSessionReqs} = State) ->
     State1 = State#state{pending_session_reqs = [From | PendingSessionReqs]},
-    {keep_state, State1}.
+    {keep_state, State1};
+sasl_hdr_rcvds(info, {'DOWN', MRef, process, _Pid, _},
+               #state{reader_m_ref = MRef}) ->
+    {stop, {shutdown, reader_down}}.
 
 sasl_init_sent(_EvtType, #'v1_0.sasl_outcome'{code = {ubyte, 0}},
                #state{socket = Socket} = State) ->
@@ -239,7 +242,10 @@ sasl_init_sent(_EvtType, #'v1_0.sasl_outcome'{code = {ubyte, C}},
 sasl_init_sent({call, From}, begin_session,
                #state{pending_session_reqs = PendingSessionReqs} = State) ->
     State1 = State#state{pending_session_reqs = [From | PendingSessionReqs]},
-    {keep_state, State1}.
+    {keep_state, State1};
+sasl_init_sent(info, {'DOWN', MRef, process, _Pid, _},
+               #state{reader_m_ref = MRef}) ->
+    {stop, {shutdown, reader_down}}.
 
 hdr_sent(_EvtType, {protocol_header_received, 0, 1, 0, 0}, State) ->
     case send_open(State) of
@@ -254,7 +260,10 @@ hdr_sent(_EvtType, {protocol_header_received, Protocol, Maj, Min,
 hdr_sent({call, From}, begin_session,
          #state{pending_session_reqs = PendingSessionReqs} = State) ->
     State1 = State#state{pending_session_reqs = [From | PendingSessionReqs]},
-    {keep_state, State1}.
+    {keep_state, State1};
+hdr_sent(info, {'DOWN', MRef, process, _Pid, _},
+         #state{reader_m_ref = MRef}) ->
+    {stop, {shutdown, reader_down}}.
 
 open_sent(_EvtType, #'v1_0.open'{max_frame_size = MaybeMaxFrameSize,
                                  idle_time_out = Timeout} = Open,
@@ -350,11 +359,15 @@ opened(_EvtType, Frame, State) ->
 close_sent(_EvtType, heartbeat, _Data) ->
     keep_state_and_data;
 close_sent(_EvtType, {'EXIT', _Pid, shutdown}, _Data) ->
-    %% monitored processes may exit during closure
+    %% our supervisor is shutting us down; let it enforce its own timeout
     keep_state_and_data;
 close_sent(_EvtType, {'EXIT', _Pid, {shutdown, _}}, _Data) ->
-    %% monitored processes may exit during closure
+    %% our supervisor is shutting us down; let it enforce its own timeout
     keep_state_and_data;
+close_sent(_EvtType, {'EXIT', _Pid, _Reason}, _Data) ->
+    %% a linked process exited unexpectedly; we probably won't
+    %% receive a close frame
+    {stop, normal};
 close_sent(_EvtType, {'DOWN', _Ref, process, ReaderPid, _Reason},
            #state{reader = ReaderPid}) ->
     %% if the reader exits we probably won't receive a close frame