@@ -5,7 +5,7 @@ use linkerd2_error::Error;
55use linkerd2_error_metrics as metrics;
66use linkerd2_error_respond as respond;
77pub use linkerd2_error_respond:: RespondLayer ;
8- use linkerd2_proxy_http:: HasH2Reason ;
8+ use linkerd2_proxy_http:: { client_handle :: Close , ClientHandle , HasH2Reason } ;
99use linkerd2_timeout:: { error:: ResponseTimeout , FailFastError } ;
1010use pin_project:: pin_project;
1111use std:: pin:: Pin ;
@@ -51,10 +51,11 @@ pub enum Reason {
5151#[ derive( Copy , Clone , Debug ) ]
5252pub struct NewRespond ( ( ) ) ;
5353
54- #[ derive( Copy , Clone , Debug ) ]
55- pub enum Respond {
56- Http1 ( http:: Version ) ,
57- Http2 { is_grpc : bool } ,
54+ #[ derive( Clone , Debug ) ]
55+ pub struct Respond {
56+ version : http:: Version ,
57+ is_grpc : bool ,
58+ close : Option < Close > ,
5859}
5960
6061#[ pin_project( project = ResponseBodyProj ) ]
@@ -139,30 +140,39 @@ impl<ReqB, RspB: Default + hyper::body::HttpBody>
139140 type Respond = Respond ;
140141
141142 fn new_respond ( & self , req : & http:: Request < ReqB > ) -> Self :: Respond {
143+ let close = req
144+ . extensions ( )
145+ . get :: < ClientHandle > ( )
146+ . map ( |h| h. close . clone ( ) ) ;
142147 match req. version ( ) {
143148 http:: Version :: HTTP_2 => {
144149 let is_grpc = req
145150 . headers ( )
146151 . get ( http:: header:: CONTENT_TYPE )
147152 . and_then ( |v| v. to_str ( ) . ok ( ) . map ( |s| s. starts_with ( "application/grpc" ) ) )
148153 . unwrap_or ( false ) ;
149- Respond :: Http2 { is_grpc }
154+ Respond {
155+ is_grpc,
156+ close,
157+ version : http:: Version :: HTTP_2 ,
158+ }
150159 }
151- version => Respond :: Http1 ( version) ,
160+ version => Respond {
161+ version,
162+ close,
163+ is_grpc : false ,
164+ } ,
152165 }
153166 }
154167}
155168
156169impl < RspB : Default + hyper:: body:: HttpBody > respond:: Respond < http:: Response < RspB > > for Respond {
157170 type Response = http:: Response < ResponseBody < RspB > > ;
158171
159- fn respond (
160- & self ,
161- reseponse : Result < http:: Response < RspB > , Error > ,
162- ) -> Result < Self :: Response , Error > {
163- match reseponse {
172+ fn respond ( & self , res : Result < http:: Response < RspB > , Error > ) -> Result < Self :: Response , Error > {
173+ match res {
164174 Ok ( response) => Ok ( response. map ( |b| match * self {
165- Respond :: Http2 { is_grpc } if is_grpc == true => ResponseBody :: Grpc {
175+ Respond { is_grpc : true , .. } => ResponseBody :: Grpc {
166176 inner : b,
167177 trailers : None ,
168178 } ,
@@ -171,33 +181,36 @@ impl<RspB: Default + hyper::body::HttpBody> respond::Respond<http::Response<RspB
171181 Err ( error) => {
172182 warn ! ( "Failed to proxy request: {}" , error) ;
173183
174- if let Respond :: Http2 { is_grpc } = self {
184+ if self . version == http :: Version :: HTTP_2 {
175185 if let Some ( reset) = error. h2_reason ( ) {
176186 debug ! ( %reset, "Propagating HTTP2 reset" ) ;
177187 return Err ( error) ;
178188 }
189+ }
179190
180- if * is_grpc {
181- let mut rsp = http:: Response :: builder ( )
182- . version ( http:: Version :: HTTP_2 )
183- . header ( http:: header:: CONTENT_LENGTH , "0" )
184- . body ( ResponseBody :: default ( ) )
185- . expect ( "app::errors response is valid" ) ;
186- let code = set_grpc_status ( & * error, rsp. headers_mut ( ) ) ;
187- debug ! ( ?code, "Handling error with gRPC status" ) ;
188- return Ok ( rsp) ;
191+ // Gracefully teardown the server-side connection.
192+ if should_teardown_connection ( & * error) {
193+ if let Some ( c) = self . close . as_ref ( ) {
194+ debug ! ( "Closing server-side connection" ) ;
195+ c. close ( ) ;
189196 }
190197 }
191198
192- let version = match self {
193- Respond :: Http1 ( ref version) => version. clone ( ) ,
194- Respond :: Http2 { .. } => http:: Version :: HTTP_2 ,
195- } ;
199+ if self . is_grpc {
200+ let mut rsp = http:: Response :: builder ( )
201+ . version ( http:: Version :: HTTP_2 )
202+ . header ( http:: header:: CONTENT_LENGTH , "0" )
203+ . body ( ResponseBody :: default ( ) )
204+ . expect ( "app::errors response is valid" ) ;
205+ let code = set_grpc_status ( & * error, rsp. headers_mut ( ) ) ;
206+ debug ! ( ?code, "Handling error with gRPC status" ) ;
207+ return Ok ( rsp) ;
208+ }
196209
197210 let status = http_status ( & * error) ;
198- debug ! ( %status, ? version, "Handling error with HTTP response" ) ;
211+ debug ! ( %status, version = ? self . version, "Handling error with HTTP response" ) ;
199212 Ok ( http:: Response :: builder ( )
200- . version ( version)
213+ . version ( self . version )
201214 . status ( status)
202215 . header ( http:: header:: CONTENT_LENGTH , "0" )
203216 . body ( ResponseBody :: default ( ) )
@@ -207,6 +220,18 @@ impl<RspB: Default + hyper::body::HttpBody> respond::Respond<http::Response<RspB
207220 }
208221}
209222
223+ fn should_teardown_connection ( error : & ( dyn std:: error:: Error + ' static ) ) -> bool {
224+ if error. is :: < ResponseTimeout > ( ) {
225+ false
226+ } else if error. is :: < tower:: timeout:: error:: Elapsed > ( ) {
227+ false
228+ } else if let Some ( e) = error. source ( ) {
229+ should_teardown_connection ( e)
230+ } else {
231+ true
232+ }
233+ }
234+
210235fn http_status ( error : & ( dyn std:: error:: Error + ' static ) ) -> StatusCode {
211236 if let Some ( HttpError { http, .. } ) = error. downcast_ref :: < HttpError > ( ) {
212237 * http
0 commit comments