[consensus::marshal] Mailbox functions return error.

[BrendanChou]

Ergonomic changes. consensus::marshal methods now return Result to optionally indicate a processing error (rather than flattening into a None)

[SuperFluffy]

Yeah, that's nice.

I wonder if withthese new APIs the special cases in the Automaton implementors could go away (for example alto) that always check if a digest if genesis.

[cloudflare-workers-and-pages]

Deploying monorepo with    Cloudflare Pages

Latest commit:	430a161
Status:	✅  Deploy successful!
Preview URL:	https://60334dcf.monorepo-eu0.pages.dev
Branch Preview URL:	https://bc-marshal-getters-3.monorepo-eu0.pages.dev

View logs

[BrendanChou]

I removed any of the genesis stuff from this PR to keep it scoped

Probably will go into #1707 or its own PR

[patrick-ogrady]

Suspect this will be a pattern we eventually follow in #1110

[SuperFluffy]

To move subscribe closer to the other types methods of this PR, instead of a oneshot::Receiver<B> one could return a Subscription<B> that exposes some of the same events, i.e. (typed this from memory and haven't run it; so this might not compile):

use std::pin::Pin;
use std::task::Poll;
use futures::{ready, FutureExt as _};

/// A subscription to a block returned by [`Mailbox::subscribe`].
pin_project_lite::pin_project! {
    struct Subscription<B> {
        inner: oneshot::Receiver<B>,
    }
}

impl<B> Subscription<B> {
    // recv_blocking or whatever; probably not even necessary
}

impl<B> Future for Subscription<B> {
    type Output = Result<B, Closed>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        match ready!(this.inner.poll_unpin(cx)) }
            Ok(value) -> Poll::Ready(Ok(value)),
            Err(_err) -> Poll::Ready(Err(Closed)),
        }
    }
}

[BrendanChou]

My feeling is that this would add type noise without much benefit. Would prefer the client uses map_err(|_| Closed) at the call site if they REALLY need the error type to be the same between two different calls to marshal, though I don't expect this to be needed most of the time.

To me, the other alternative I would consider is Result<oneshot::Receiver<B>, Closed> in order to propagate the Closed error on send() directly, but marshal should never drop the subscription unless the actor itself shuts down, so it doesn't seem necessary to me to differentiate between Closed and oneshot::Canceled

[SuperFluffy]

Yeah, I think I agree.

I also think that returning the explicit subscription feels unnecessary (just like it was for the other cases). Say somebody really wanted to send the subscription to another thread to be awaited there instead of at the call site, they could just move the mailbox to an async fn:

# Current situation

let subscription = mailbox.subscribe().await;
# do something else
let block = subscription.await;

# await the oneshot inside the `fn subscribe`
let subscription = { let mailbox = mailbox.clone(); async move { mailbox().subscribe().await } };
# do something else
let block = subscription.await;

Now if you wanted to dot get rid of the boilerplate (because moving the subscription is a reaistic use case), a common pattern is:

async fn subscribe_owned(self) -> Block {}

# usage
let subscription = mailbox.clone().subscribe_owned();

# do smth else
let block = subscription.await;

[BrendanChou]

Returning the explicit oneshot::Receiver<B> seems like a feature to me in this case, avoids holding &mut self across awaiting for the block to become available. We would expect the caller move the await elsewhere, for example if they want to put it in a select! opposite a timeout.

As you mentioned, cloning the mailbox first would work, but it could be easy to overlook for clients (easy to forget). I think it's more ergonomic as-is. If we were to add a more ergonomic version for awaiting the block, I suggest cloning the sender inside of that function, which removes the need for the client to clone the mailbox. For example:

pub async fn wait_for_block(
    &self, // avoids borrowing `&mut self`
    round: Option<Round>,
    commitment: B::Commitment,
) -> Result<B, Closed> {
    let (tx, rx) = oneshot::channel();
    let mut sender = self.sender.clone();
    sender
        .send(Message::Subscribe { round, commitment, response: tx })
        .await
        .map_err(|_| Closed)?;
    rx.await.map_err(|_| Closed)
}

[SuperFluffy]

That's not quite the same because async fn f(&self) -> T desugars to fn f(&self) -> impl Future<Output = T> + 'a, and so you don't get 'static future, irrespective of the function body. So client.clone().f_owned() is the go-to for functions for futures that need mut self and have cheaply clonable clients.

I guess you could do fn f(&self) -> impl Future<Output = T> + 'static.

[BrendanChou]

Looks like you're suggesting the following?

    use core::future::Future;
...
    pub fn subscribe_owned(
        &self,
        round: Option<Round>,
        commitment: B::Commitment,
    ) -> impl Future<Output = Result<B, Closed>> + 'static {
        let mut sender = self.sender.clone();
        async move {
            let (tx, rx) = oneshot::channel();
            sender
                .send(Message::Subscribe {
                    round,
                    commitment,
                    response: tx,
                })
                .await
                .map_err(|_| Closed)?;
            rx.await.map_err(|_| Closed)
        }
    }

And use it by:

let subscriber = marshal.clone();
let subscription = subscriber.subscribe_owned(...);
...
let block = subscription.await?;

[SuperFluffy]

Yeah, that's what I was thinking of. Although it's less nice to look at when looking at docs (I know that I usually shy away from scary looking signatures).

For what it's worth, there are examples in the ecosystem for either introducing async fn f_owned methods [1] [2], or simply telling people to clone their clients clients to work around concurrent usage of &mut self methods [3] (just a few that I remember at the top of my head).

1: https://docs.rs/tokio-util/latest/tokio_util/sync/struct.CancellationToken.html#method.cancelled_owned
2: https://docs.rs/tokio/latest/tokio/sync/mpsc/struct.Sender.html#method.try_reserve_owned
3: https://docs.rs/tonic/latest/tonic/client/index.html#concurrent-usage

[codecov]

Codecov Report

❌ Patch coverage is 93.78882% with 10 lines in your changes missing coverage. Please review.
✅ Project coverage is 92.20%. Comparing base (87fff07) to head (430a161).
⚠️ Report is 7 commits behind head on main.

Files with missing lines	Patch %	Lines
consensus/src/marshal/mocks/resolver.rs	0.00%	10 Missing ⚠️

@@            Coverage Diff             @@
##             main    #1745      +/-   ##
==========================================
- Coverage   92.28%   92.20%   -0.09%     
==========================================
  Files         304      305       +1     
  Lines       79179    79528     +349     
==========================================
+ Hits        73072    73327     +255     
- Misses       6107     6201      +94     

Files with missing lines	Coverage Δ
consensus/src/marshal/actor.rs	89.37% <100.00%> (+0.66%)	⬆️
consensus/src/marshal/ingress/mailbox.rs	91.13% <100.00%> (+7.13%)	⬆️
consensus/src/marshal/mocks/application.rs	100.00% <100.00%> (ø)
consensus/src/marshal/mod.rs	99.84% <100.00%> (+0.02%)	⬆️
consensus/src/marshal/mocks/resolver.rs	0.00% <0.00%> (ø)

... and 59 files with indirect coverage changes

---

Continue to review full report in Codecov by Sentry.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 87fff07...430a161. Read the comment docs.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.