Is there an alternative pattern for OutputPort for reliable message delivery?

[stuartcarnie]

We've encountered a challenge when using an OutputPort in an actor that may be stopped. OutputPort.send message delivery occurs asynchronously, but is dependent on the lifetime of OutputPort.

For example:

1. Actor A state contains a subscribers: OutputPort<T> field
2. Actor A broadcasts a message to the subscribers via the send API
3. Actor A then stops, which causes Actor A, and it's state to be dropped

If we look at the implementation of OutputPort<T>, we see that Drop stops the subscribers:

ractor/ractor/src/port/output/mod.rs

Lines 108 to 112 in 1949d92

	let mut subs = self.subscriptions.write().unwrap();
	for sub in subs.iter_mut() {
	sub.stop();
	}
	subs.clear();

The implementation of stop aborts the JoinHandle for the subscriber, and therefore the associated task is aborted immediately:

ractor/ractor/src/port/output/mod.rs

Lines 131 to 134 in 1949d92

	/// Stop the subscription, by aborting the underlying [JoinHandle]
	pub(crate) fn stop(&mut self) {
	self.handle.abort();
	}

which we see is a separate task that was spawned when the actor subscribed:

ractor/ractor/src/port/output/mod.rs

Lines 147 to 156 in 1949d92

	let handle = crate::concurrency::spawn(async move {
	while let Ok(Some(msg)) = port.recv().await {
	if let Some(new_msg) = converter(msg) {
	if receiver.cast(new_msg).is_err() {
	// kill the subscription process, as the forwarding agent is stopped
	return;
	}
	}
	}
	});

which is responsible for sending the message.

[stuartcarnie]

I've looked a bit closer, and we could eliminate the implementation of Drop on the OutputPort type, as the receivers will all terminate immediately after draining the broadcast channel anyway, because the tx side of the broadcast channel will be dropped.

[stuartcarnie]

Submitted a PR to resolve the issue:

• fix: Ensure pending messages are delivered when OutputPort is dropped. #293