Repository: actix/actix (language: rust)

Issue:
Actor stops receiving messages after filling the mailbox
The following code demonstrates the bug. With actix 0.12.0 + actix-broker 1.4.1 the program prints `handle 0` to `handle 99`. With actix 0.13.0 + actix-broker 1.4.2 we only get `handle 0` to `handle 15` because after reaching mailbox capacity (16 by default) the async message subscriber gets detached from the broker.

```rust
use actix::*;
use actix_broker::*;
use std::time::Duration;

struct MyActor;

#[derive(Message, Clone)]
#[rtype("()")]
struct MyMsg(u32);

impl Actor for MyActor {
    type Context = Context<Self>;
    fn started(&mut self, ctx: &mut Self::Context) {
        // ctx.set_mailbox_capacity(1_000_000_000);
        self.subscribe_async::<SystemBroker, MyMsg>(ctx);
    }
}

impl Handler<MyMsg> for MyActor {
    type Result = ();

    fn handle(&mut self, msg: MyMsg, _ctx: &mut Self::Context) -> Self::Result {
        println!("handle {}", msg.0);
    }
}

#[actix_rt::main]
async fn test() {
    let actor = MyActor;
    let _addr = actor.start();

    // give the actor time to call `started` method
    tokio::time::sleep(Duration::from_secs(1)).await;

    for i in 0..50 {
        Broker::<SystemBroker>::issue_async(MyMsg(i));
    }

    // let the actor handle all messages we sent so far
    tokio::time::sleep(Duration::from_secs(1)).await;

    for i in 50..100 {
        Broker::<SystemBroker>::issue_async(MyMsg(i));
    }
   
    // let the actor handle the second batch of messages
    tokio::time::sleep(Duration::from_secs(1)).await;
}
```

## Expected Behavior
- The actor handles all messages - 0 to 99

## Current Behavior
- Broker delivers messages until the mailbox if full
- When mailbox is full, messages are discarded
- New messages don't get delivered to the actor even when the mailbox is emptied

## Possible Solution
- You could increase size of the mailbox
- I suspect the bug was introduced here: https://github.com/actix/actix/commit/2bcb72ca267a11d9cb1e16b9f05f6bbc4174f917#diff-18ef0c609b5c921ad3b82c8a613aafce763d0458c0a945f2f0e45dee49022742R131. We could revert back to `do_send`. It inserts messages even when the mailbox is full, which could be considered a bug too, but seems to be less harmful than detaching a subscriber.

## Steps to Reproduce (for bugs)
1. Create an actor with a small mailbox
2. Subscribe for some message with SystemBroker using `subscribe_async`
3. Emit at one mailbox worth of messages on the SystemBroker using `issue_async`
4. Emit some more messages after that and see if they get handled or not

## Context
I discovered the issue when a web service I have recently updated to actix 0.13.0 suddenly stopped forwarding data to other services. It did that after reaching HTTP traffic of around 100 requests per second.

## Your Environment

* Rust version (I.e, output of `rustc -V`): 1.58.1
* actix version: 0.13.0
* actix-broker version: 0.4.2



Discussion / hints:
I don't think that reverting to `do_send` is the right solution since it means the actor can't apply backpressure.
The broker should probably have an async API to allow the caller to wait for delivery to all subscribers and an API to try to deliver, though this is problematic if an actor is both the sender and the listener of the message. Maybe we internals of the broker should await the send with a configurable timeout/retry/backoff.

Fix patch:
diff --git a/actix-broker/CHANGES.md b/actix-broker/CHANGES.md
index fcec3cbf3..d8d36ff6a 100644
--- a/actix-broker/CHANGES.md
+++ b/actix-broker/CHANGES.md
@@ -1,7 +1,9 @@
 # Changes
 
-## Unreleased - 2021-xx-xx
+## Unreleased - 2022-xx-xx
+- Fix at-least-once delivery guarantee for IssueAsync. [#536]
 
+[#536]: https://github.com/actix/actix/pull/536
 
 ## 0.4.2 - 2022-03-01
 - Add support for `actix` v0.13.
diff --git a/actix-broker/src/broker.rs b/actix-broker/src/broker.rs
index 80716d8da..3031f9b1e 100644
--- a/actix-broker/src/broker.rs
+++ b/actix-broker/src/broker.rs
@@ -126,15 +126,22 @@ impl<T: 'static + Unpin, M: BrokerMsg> Handler<IssueAsync<M>> for Broker<T> {
     fn handle(&mut self, msg: IssueAsync<M>, _ctx: &mut Context<Self>) {
         trace!("Broker: Received IssueAsync");
         if let Some(mut subs) = self.take_subs::<M>() {
-            subs.drain(..)
-                .filter_map(|(id, s)| {
-                    if id == msg.1 || s.try_send(msg.0.clone()).is_ok() {
-                        Some((id, s))
-                    } else {
-                        None
+            subs.drain(..).for_each(|(id, s)| {
+                if id == msg.1 {
+                    self.add_sub::<M>(s, id);
+                } else {
+                    match s.try_send(msg.0.clone()) {
+                        Ok(_) => self.add_sub::<M>(s, id),
+                        Err(SendError::Full(_)) => {
+                            // Ensure that that the message is delivered even if the mailbox is full.
+                            // We do a try first to remove receiver that have closed their mailbox.
+                            s.do_send(msg.0.clone());
+                            self.add_sub::<M>(s, id);
+                        }
+                        Err(_) => (),
                     }
-                })
-                .for_each(|(id, s)| self.add_sub::<M>(s, id));
+                }
+            });
         }
         self.set_msg::<M>(msg.0);
     }
diff --git a/actix-broker/src/issue.rs b/actix-broker/src/issue.rs
index 8bc9fc3f5..5dfcc1b2a 100644
--- a/actix-broker/src/issue.rs
+++ b/actix-broker/src/issue.rs
@@ -14,6 +14,9 @@ where
     <Self as Actor>::Context: AsyncContext<Self>,
 {
     /// Asynchronously issue a message.
+    /// This bypasses the mailbox capacity, and  will always queue the message.
+    /// If the mailbox is closed, the message is silently dropped and the subscriber
+    /// is detached from the broker.
     fn issue_async<T: RegisteredBroker, M: BrokerMsg>(&self, msg: M) {
         let broker = T::get_broker();
         broker.do_send(IssueAsync(msg, TypeId::of::<Self>()));


Test patch:
diff --git a/actix-broker/tests/delivery.rs b/actix-broker/tests/delivery.rs
new file mode 100644
index 000000000..620cf8dd7
--- /dev/null
+++ b/actix-broker/tests/delivery.rs
@@ -0,0 +1,65 @@
+extern crate actix;
+extern crate actix_broker;
+
+use actix::clock::sleep;
+use actix::prelude::*;
+use actix_broker::{Broker, BrokerSubscribe, SystemBroker};
+
+use std::time::Duration;
+
+#[derive(Clone, Message)]
+#[rtype(result = "()")]
+struct TestMessage;
+
+#[derive(Default)]
+struct TestActor {
+    count: u8,
+}
+
+impl Actor for TestActor {
+    type Context = Context<Self>;
+
+    fn started(&mut self, ctx: &mut Self::Context) {
+        self.subscribe_async::<SystemBroker, TestMessage>(ctx);
+    }
+}
+
+impl Handler<TestMessage> for TestActor {
+    type Result = AtomicResponse<Self, ()>;
+
+    fn handle(&mut self, _msg: TestMessage, _ctx: &mut Self::Context) -> Self::Result {
+        let first = self.count == 0;
+        let task = async move {
+            if first {
+                sleep(Duration::from_millis(500)).await;
+            }
+        }
+        .into_actor(self)
+        .map(|_, act, _ctx| {
+            act.count += 1;
+            if act.count == 51 {
+                System::current().stop();
+            }
+        })
+        .boxed_local();
+
+        AtomicResponse::new(task)
+    }
+}
+
+#[test]
+fn it_issues_async_on_full_mailbox() {
+    let sys = System::new();
+
+    sys.block_on(async {
+        let addr = TestActor::default().start();
+        sleep(Duration::from_millis(100)).await;
+        addr.try_send(TestMessage)
+            .expect("Unable to send base message");
+        for _ in 0..50 {
+            Broker::<SystemBroker>::issue_async(TestMessage);
+        }
+    });
+
+    sys.run().unwrap();
+}