Discard request stream when server dispatcher receives reset (linkerd…

…#2264)

When Linkerd's ServerDispatcher receives a stream reset immediately after receiving a request, it `raise`s to cancel sending that request to the service.  In this case, the request is dropped but may have unread data frames in its stream.  This can result in a direct memory leak because these unread frames retain a non-zero reference count even though they are unreachable.  

When `raise`ing the send of the request, we also cancel the request's stream which properly releases any frames in the stream.

Signed-off-by: Alex Leong <alex@buoyant.io>

finagle/h2/src/main/scala/com/twitter/finagle/buoyant/h2/netty4/Netty4ServerDispatcher.scala

@@ -85,11 +85,17 @@ class Netty4ServerDispatcher(
     val serveF = st.onRecvMessage.flatMap(serve).flatMap(st.send(_).flatten)
 
     // When the stream is reset, ensure that the cancelation is
-    // propagated downstream.
+    // propagated downstream and discard the recv message (if we have one).
     st.onReset.onFailure {
-      case StreamError.Remote(rst: Reset) => serveF.raise(rst)
-      case StreamError.Remote(e) => serveF.raise(Reset.Cancel)
-      case e => serveF.raise(e)
+      case StreamError.Remote(rst: Reset) =>
+        st.onRecvMessage.onSuccess(_.stream.cancel(rst))
+        serveF.raise(rst)
+      case StreamError.Remote(e) =>
+        st.onRecvMessage.onSuccess(_.stream.cancel(Reset.Cancel))
+        serveF.raise(Reset.Cancel)
+      case e =>
+        st.onRecvMessage.onSuccess(_.stream.cancel(Reset.Cancel))
+        serveF.raise(e)
     }
 
     val _ = serveF.onFailure {