Implemented Filter::then

[gtsiam]

Tipping point for me to jump into Future type hell to implement this was this comment - but I'd say it's worth it. I've used it in my personal project for a while, and the code I've been able to write is much cleaner, I think.

Changes:

• Added Filter::then() and complementary type Then. Exactly like map, but async. The name is meant to reflect future's then() function.
• Filter::and_then() documentation now points to Filter::then() as an alternative, since my understanding is that:

  Rejections are meant to say "this filter didn't accept the request, maybe another can."

Fix #712
Fix #594

Providing Reply implementations for Result would also be very useful, so this is loosely related to #458

---

Post-edit note: This function was renamed from map_async to then to more closely match future's api. However it is worth noting that just as Filter::map isn't exactly like future's map on account of rejections, Filter::then isn't exactly like future's then. However Filter::then is the same, conceptually, as future's then, so this seems like the best choice.

[apiraino]

hi @gtsiam I was called from the issue you linked, I'm really interested in understanding your PR and how in practice could be used. Please fix my understanding.

By reading again Sean's comment my hunch is that your PR is trying to draw a clearer line between a Rejection (i.e. "current Filter cannot handle this request, goto next one or fail with a warp::reject::not_found") and an actual HTTP error code I want to return to the client (possibly bypassing all remaining filters?). All to avoid the current (imho) hack of abusing the Rejection type by adding custom rejections and then using .recover() on them to emit the correct reply.

Can you please show here an example of how to use your map_async returning a Future that (I'm assuming your intent) will be converted into a Reply?

Regardless, thank you from me for working on this.

[gtsiam]

Pretty much, yes.

map_async is exactly like map, only async. You still have to map your error, if it isn't already, into something impl Reply. for example I have something like this:

async fn login(state: State, req: LoginRequest) -> Result<LoginResponse, LoginError> { /* ... */ }
fn json_response<T: Serialize, E: Serialize>(res: Result<T, E>) -> reply::Json { /* ... */ }

// In my filter function:
path!("login")
    .and(post())
    .and(with_state)
    .and(body::json())
    .map_async(login)
    .map(json_response);

Notice that json_response takes in a Result<T, E>, not just T. Also nothing compels me to use result here, other than the fact that my code calls for it. Just like map, map_async doesn't care what you return. It only forwards it to the next filter.

Warp's api today looks like this:

Closure returns	Function
T	map()
Result<T, Rejection>	???
Future<Output = T>	???
Future<Output = Result<T, Rejection>>	and_then()

The idea is to fill in the second gap with map_async() so that you don't have to deal with rejections for application-level errors in async code. Arguably, there are two gaps, but I doubt a sync version of and_then() is really worth the effort.

This would make the transition from sync to async as simple as changing your function to return impl Future and using map_async instead of map.

Then there's the ? operator. Imagine, for example, you have to make a call to a database using async:

#[derive(Debug, From, Serialize)]
enum LoginError {
    // Other variants here. InternalError is a catch-all for, well, internal errors.
    Internal(#[from(forward)] api::InternalError),
}

// No need for impl Reject for LoginError. Warp never deconstructs the returned Result.

async fn login(state: State, req: LoginRequest) -> Result<LoginResponse, LoginError> {
    let conn = state.pool.get()?; // No need to map_err(|err| reject::custom(err)) !
    
    // [...] Stuff that doesn't really matter for this example
}

You are already able to do just this with map(), but only for sync code.

There's also infallible async endpoints where you have to do stuff like this because of type inference:

.and_then(|data| async move {
        let response = data.get().await;
        Ok::<_, Rejection>(warp::reply::json(&response))
    });

But with map_async:

.map_async(|data| async move {
        let response = data.get().await;
        warp::reply::json(&response)
    });

Also a fundamental issue with using Rejection for application errors is that if you forget to handle an error in .recover(), then warp has to check all other handlers. Where are rust's compile-time guarantees?

I'd wager there's also some overhead to using recover to check for a lot of errors, though I could be wrong. Rejection is not magic. It is, essentially, a linked list of errors. Every find has to go through it (though I'm not sure how long it grows in practice, I'd have to check when I have time for it)

(Also while writing, I just realized this probably closes #594 too - I'll edit the description)

[apiraino]

Fantastic explaination, thank you so much. You clarified what I hoped for (i.e. remove all the custom Rejections which are really confusing).

Also a fundamental issue with using Rejection for application errors is that if you forget to handle an error in .recover(), then warp has to check all other handlers. Where are rust's compile-time guarantees?

Also, I suspect, this could potentially solve another annoying side-effect: if a request goes through the chain of filters and reaches the end without being "picked up" and handled I could avoid the MethodNotAllowed pile-up (see this comment for example) that got me a couple of times.

EDIT:

I'd wager there's also some overhead to using recover to check for a lot of errors, ... Rejection is not magic. It is, essentially, a linked list of errors.

well, basically what you just said :-)

[lkts]

This is a great addition. I have just followed the same trap - using and_then for async handler, type signature suggests my errors should map to Rejection, i try to implement recover and start to question myself if it's the right thing to do and apparently it is not.

[kaj]

I can't say I understand all the details of implementation and pinning of the future, but it looks good to me, and seems to be similar to and different from the and_then implementation in the right ways. I like the doc changes to Filter::and_then.

The changes is src/filter/and_then.rs is not really relevant to implementing map_async, but nice cleanup.

[gtsiam]

Good to hear :)

I ended up with this after copy-pasting the and_then implementation, renaming it, then incrementally fixing it until I ended up with something that resembled the map_async I wanted - then came the fun part of trying to reason through it to make sure it made sense - so yeah... point is; should be very similar.

As for src/filter/and_then.rs, entirely my own pet-peeve what can I say.

[gtsiam]

I was hoping the convinient "Update" branch button would just rebase it... guess not. Let me do it manually then.

[gtsiam]

Actually it just occurred to me github could probably merge that anyway and rebasing messes with reviews... so uh.... oops?
Well, not much I can do about it now, but the changes are the same.

[kaj]

My review still stands, but I'm no maintainer ...

[apiraino]

@seanmonstar @jxs gently bumping for a review of this PR

thanks!

[option-greek]

Hello all,
If I have to use this feature (map_async), which version of warp should I be pulling in from crates.io?

[apiraino]

@option-greek this is a branch from a fork waiting for a review from the maintainers. As such, it is not published anywhere. If you want to test it, you need to checkout this fork and use it in your project with something like:

warp = { git="https://github.com/gtsiam/warp.git", rev="b16c88" }

(I didn't test the above import)

[option-greek]

Got it. I'll try it out and post here the results. Btw, didn't realise import directly from github and that too from a commit was possible!

…

On Mon, 6 Sep 2021, 21:00 apiraino, ***@***.***> wrote: @option-greek <https://github.com/option-greek> this is a branch from a fork waiting for a review from the maintainers. As such, it is not published anywhere. If you want to test it, you need to checkout this fork and use it in your project with something like: warp = { git="https://github.com/gtsiam/warp.git", rev="b16c88" } (I didn't test the above import) — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#878 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAK5GI2M2PF6PLKDPLZAKN3UATM7ZANCNFSM4723CFJA> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>.

[rubycut]

Any ETA for this PR?

[gtsiam]

🎉

@rubycut ETA: -3m

[beingflo]

I have come across this PR after researching my error handling problem in warp. After trying to get then working I'm still quite confused and hope someone can clarify for me :)

Previously I've fallen in the same trap of converting my DB errors into Rejections and handling them in the recover which is quite unwieldy. Previously, I had something like the following:

let login = warp::post()
        .and(warp::path("login"))
        .and(warp::body::json())
        .and(with_db.clone())
        .and_then(user::login);

pub async fn login(user: UserCredentials, db: PgPool) -> Result<impl warp::Reply, warp::Rejection> {
    if !user_exists(&user.name, &db).await? {
        return Ok(StatusCode::UNAUTHORIZED.into_response());
    }
   ...
}

The user_exists function returns an ApiError enum that among others, contains the DBError, further the ApiError implements reject.

Is there any way I could make the use of Rejection go away and still keep the handy .await?. I figured I could use then and have the return value impl::warp::Reply and implement Reply for ApiError, but that throws the trait From<ApiError> is not implemented for warp::http::Error errors.

Could someone point out to me what I'm missing? It's getting quite frustrating spending the majority of my time with error handling and not business logic :)

[gtsiam]

Off the top of my head, I'd say to try something like this:

let login = warp::post()
        .and(warp::path("login"))
        .and(warp::body::json())
        .and(with_db.clone())
        .then(user::login)          // Note that then() doesn't care about Reply / Rejection
        .map(util::result_reply);   // What's important is that whatever comes out here, implements Reply

pub async fn login(user: UserCredentials, db: PgPool) -> Result<impl warp::Reply, ApiError> {
    if !user_exists(&user.name, &db).await? {
        return Ok(StatusCode::UNAUTHORIZED.into_response());
    }
   ...
}

So, assuming you implement Reply for ApiError, util::result_reply could be something along the lines of:

pub async fn result_reply<T: Reply, E: Reply>(res: Result<T, E>) -> impl Reply {
    match res {
        Ok(r) => r.into_response(),
        Err(e) => e.into_response(),
    }
}

---

This last part (result_reply) would be unnecessary if we had an impl Reply for Result<impl Reply, impl Reply>, but that's a question for another pr - Which I'm probably gonna do now that I think about it, since it's a small quality of life change. Originally I was counting on #458 getting merged, but I just found out that pr also implemented lots of other things, including then (formerly map_async), so it got closed.

[beingflo]

Now it makes sense, thanks a lot :). This should make my code quite a bit cleaner. Looking forward to have a v0.3.2 with this change in it to avoid git dependency.

[jakajancar]

This is great to have warp return an error from a handler and be done with it, instead of continuing to try a bunch of other filters.

However, how should one handle various "extractor" filters that are fallible? For example, if you have an .and(user(db)) filter that internally calls e.g. warp::header() but then connects to the database which might fail. If such a filter just rejects, other peer filters will still be attempted (perhaps making the very same query 10 times). And you cannot just map to impl Reply because you're actually extracting some value that you want to use later.

In this case, "fatal rejections" would still be needed to avoid unnecessary processing, no?

[jakajancar]

@gtsiam bringing the above to your attention, now that there’s renewed “activity” :) any ideas?

[gtsiam]

Let me experiment a bit and I'll draft a few proposals in the coming days. In the meantime, please spin up a new issue for this.

[jakajancar]

Added: #1053