Repository: actix/actix (language: rust)

Issue:
ctx.wait() does not stop processing other events sometimes
## Expected Behavior
`ctx.wait()` should always "block" actor, so it wont process any other events until waited future is resolved

## Current Behavior
`ctx.wait()` does not stop actor from processing other events sometimes, see code sample.

## Possible Solution
¯\_(ツ)_/¯

## Steps to Reproduce (for bugs)
My minimal reproducible example:
```rust
use actix::prelude::*;
use actix_rt::time::sleep;
use futures::StreamExt as _;

struct MyActor {
    wait_finished: bool,
}

impl Actor for MyActor {
    type Context = Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        ctx.add_stream(futures::stream::iter(vec![(),(),(),()]));
    }
}

impl StreamHandler<()> for MyActor {
    fn handle(&mut self, _: (), ctx: &mut Self::Context) {
        assert!(self.wait_finished);
        self.wait_finished = false;

        ctx.wait(async {}.into_actor(self).map(|_,this,ctx| {
            let handle = ctx.run_later(std::time::Duration::from_secs(1), |_, _| {});
            ctx.cancel_future(handle);

            this.wait_finished = true;
        }));
    }
}

#[actix_rt::main]
async fn main() {
    MyActor { wait_finished: true }.start();
    sleep(std::time::Duration::from_secs(1)).await;
}
```

It is expected that this should not panic, since `handle` is supposed to run only when waited future is done.

## Your Environment

* Rust Version (I.e, output of `rustc -V`): `1.49`
* Actix Version: current master (`0.11.0-beta.1`)



Discussion / hints:

Fix patch:
diff --git a/actix/src/contextimpl.rs b/actix/src/contextimpl.rs
index fcbc9d792..4b8e8cebb 100644
--- a/actix/src/contextimpl.rs
+++ b/actix/src/contextimpl.rs
@@ -4,6 +4,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use bitflags::bitflags;
+use futures_core::ready;
 use smallvec::SmallVec;
 
 use crate::actor::{
@@ -355,12 +356,8 @@ where
             // and we always have to check most recent future
             while !this.wait.is_empty() && !this.stopping() {
                 let idx = this.wait.len() - 1;
-                if let Some(item) = this.wait.last_mut() {
-                    match Pin::new(item).poll(&mut this.act, &mut this.ctx, cx) {
-                        Poll::Ready(_) => (),
-                        Poll::Pending => return Poll::Pending,
-                    }
-                }
+                let item = this.wait.last_mut().unwrap();
+                ready!(Pin::new(item).poll(&mut this.act, &mut this.ctx, cx));
                 this.wait.remove(idx);
                 this.merge();
             }
@@ -377,6 +374,11 @@ where
                 this.ctx.parts().handles[1] = this.items[idx].0;
                 match Pin::new(&mut this.items[idx].1).poll(&mut this.act, &mut this.ctx, cx) {
                     Poll::Pending => {
+                        // got new waiting item. merge
+                        if this.ctx.waiting() {
+                            this.merge();
+                        }
+
                         // check cancelled handles
                         if this.ctx.parts().handles.len() > 2 {
                             // this code is not very efficient, relaying on fact that
@@ -404,6 +406,12 @@ where
                     }
                     Poll::Ready(()) => {
                         this.items.swap_remove(idx);
+
+                        // got new waiting item. merge
+                        if this.ctx.waiting() {
+                            this.merge();
+                        }
+
                         // one of the items scheduled wait future
                         if !this.wait.is_empty() && !this.stopping() {
                             continue 'outer;


Test patch: