Attempt to fix tests

Signed-off-by: Zahari Dichev <zaharidichev@gmail.com>

linkerd/app/integration/src/controller.rs

@@ -276,13 +276,16 @@ impl pb::destination_server::Destination for Controller {
         req: grpc::Request<pb::GetDestination>,
     ) -> Result<grpc::Response<Self::GetProfileStream>, grpc::Status> {
         if let Ok(mut calls) = self.expect_profile_calls.lock() {
-            if let Some((dst, profile)) = calls.pop_front() {
-                if &dst == req.get_ref() {
-                    return Ok(grpc::Response::new(Box::pin(profile.map(Ok))));
-                }
+            match calls.pop_front() {
+                Some((dst, profile)) => {
+                    if &dst == req.get_ref() {
+                        return Ok(grpc::Response::new(Box::pin(profile.map(Ok))));
+                    }
 
-                calls.push_front((dst, profile));
-                return Err(grpc_unexpected_request());
+                    calls.push_front((dst, profile));
+                    return Err(grpc_unexpected_request());
+                }
+                None => return Ok(grpc::Response::new(Box::pin(futures::stream::empty()))),
             }
         }
 

linkerd/proxy/discover/src/make_endpoint.rs

@@ -159,7 +159,10 @@ where
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Change<D::Key, E::Response>, Error>>> {
         if let Poll::Ready(key) = self.poll_removals(cx) {
-            return Poll::Ready(Some(Ok(Change::Remove(key?))));
+            return match key {
+                Some(k) => Poll::Ready(Some(Ok(Change::Remove(k?)))),
+                None => Poll::Ready(None),
+            };
         }
 
         if let Poll::Ready(Some(res)) = self.project().make_futures.poll_next(cx) {
@@ -182,32 +185,31 @@ where
     fn poll_removals(
         self: &mut Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Result<D::Key, Error>> {
+    ) -> Poll<Option<Result<D::Key, Error>>> {
         loop {
             let mut this = self.as_mut().project();
             if let Some(key) = this.pending_removals.pop() {
                 this.make_futures.remove(&key);
-                return Poll::Ready(Ok(key));
+                return Poll::Ready(Some(Ok(key)));
             }
 
             // Before polling the resolution, where we could potentially receive
             // an `Add`, poll_ready to ensure that `make` is ready to build new
             // services. Don't process any updates until we can do so.
             ready!(this.make_endpoint.poll_ready(cx)).map_err(Into::into)?;
 
-            match ready!(this.discover.poll_discover(cx))
-                .expect("XXX(eliza): can this ever be none???")
-                .map_err(Into::into)?
-            {
-                Change::Insert(key, target) => {
+            match ready!(this.discover.poll_discover(cx)) {
+                Some(Ok(Change::Insert(key, target))) => {
                     // Start building the service and continue. If a pending
                     // service exists for this addr, it will be canceled.
                     let fut = this.make_endpoint.call(target);
                     this.make_futures.push(key, fut);
                 }
-                Change::Remove(key) => {
+                Some(Ok(Change::Remove(key))) => {
                     this.pending_removals.push(key);
                 }
+                Some(Err(err)) => return Poll::Ready(Some(Err(err.into()))),
+                None => return Poll::Ready(None),
             }
         }
     }