axum vs stream

[cemoktra]

Hi,

i was trying to stream results in an axum (0.4.4) handler but ran into problems with the Futures:

   = help: the trait `std::marker::Send` is not implemented for `dyn futures::Future<Output = Result<QueryStream, sea_orm::DbErr>>`

[billy1624]

Hey @cemoktra, can you show me a code snippet of it?

[billy1624]

I'm not familiar with Axum and can't manage to reproduce it.

async fn delete_post(
    Extension(ref conn): Extension<DatabaseConnection>,
    Path(id): Path<i32>,
    mut cookies: Cookies,
) -> Result<PostResponse, (StatusCode, &'static str)> {
    let post: post::ActiveModel = Post::find_by_id(id)
        .one(conn)
        .await
        .unwrap()
        .unwrap()
        .into();

    post.delete(conn).await.unwrap();

    let data = FlashData {
        kind: "success".to_owned(),
        message: "Post succcessfully deleted".to_owned(),
    };

    let mut stream = Post::find().stream(conn).await.unwrap();

    let mut result = Vec::new();
    while let Some(row) = stream.try_next().await.unwrap() {
        result.push(row);
    }

    Ok(post_response(&mut cookies, data))
}

sea-orm/examples/axum_example on  migration [$!] is 📦 vsea-orm-axum-example:0.1.0 via 𝗥 v1.58.0 via 🅒 base
➜ cargo b
   Compiling sea-orm-axum-example v0.1.0 (/Applications/MAMP/htdocs/sea-orm/examples/axum_example)
error[E0277]: the trait bound `fn(Extension<sea_orm::DatabaseConnection>, axum::extract::Path<i32>, Cookies) -> impl Future<Output = Result<(axum::http::StatusCode, HeaderMap), (axum::http::StatusCode, &'static str)>> {delete_post}: Handler<_, _>` is not satisfied
   --> src/main.rs:47:36
    |
47  |         .route("/delete/:id", post(delete_post))
    |                               ---- ^^^^^^^^^^^ the trait `Handler<_, _>` is not implemented for `fn(Extension<sea_orm::DatabaseConnection>, axum::extract::Path<i32>, Cookies) -> impl Future<Output = Result<(axum::http::StatusCode, HeaderMap), (axum::http::StatusCode, &'static str)>> {delete_post}`
    |                               |
    |                               required by a bound introduced by this call
    |
note: required by a bound in `post`
   --> /Users/billy/.cargo/registry/src/github.com-1ecc6299db9ec823/axum-0.4.4/src/routing/method_routing.rs:397:1
    |
397 | top_level_handler_fn!(post, POST);
    | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `post`
    = note: this error originates in the macro `top_level_handler_fn` (in Nightly builds, run with -Z macro-backtrace for more info)

For more information about this error, try `rustc --explain E0277`.
error: could not compile `sea-orm-axum-example` due to previous error

[cemoktra]

This is the exact same error. If you use axum-debug crate and annotate the handler with #[axum_debug::debug_handler] it will show you that Send is missing. And i guess it is due to impl !Send for QueryStream

[cemoktra]

Then the axum example prints when cpmpiling:

   Compiling sea-orm-axum-example v0.1.0 (/home/bastian/code/github/sea-orm/examples/axum_example)
error[E0277]: the trait bound `fn(Extension<sea_orm::DatabaseConnection>, axum::extract::Path<i32>, Cookies) -> impl futures::Future<Output = Result<(axum::http::StatusCode, HeaderMap), (axum::http::StatusCode, &'static str)>> {delete_post}: Handler<_, _>` is not satisfied
   --> src/main.rs:47:36
    |
47  |         .route("/delete/:id", post(delete_post))
    |                               ---- ^^^^^^^^^^^ the trait `Handler<_, _>` is not implemented for `fn(Extension<sea_orm::DatabaseConnection>, axum::extract::Path<i32>, Cookies) -> impl futures::Future<Output = Result<(axum::http::StatusCode, HeaderMap), (axum::http::StatusCode, &'static str)>> {delete_post}`
    |                               |
    |                               required by a bound introduced by this call
    |
note: required by a bound in `post`
   --> /home/bastian/.cargo/registry/src/github.com-1ecc6299db9ec823/axum-0.4.4/src/routing/method_routing.rs:397:1
    |
397 | top_level_handler_fn!(post, POST);
    | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `post`
    = note: this error originates in the macro `top_level_handler_fn` (in Nightly builds, run with -Z macro-backtrace for more info)

error: future cannot be sent between threads safely
   --> src/main.rs:202:1
    |
202 | async fn delete_post(
    | ^^^^^ future returned by `delete_post` is not `Send`
    |
    = help: the trait `std::marker::Send` is not implemented for `dyn futures::Stream<Item = Result<post::Model, sea_orm::DbErr>>`
note: future is not `Send` as this value is used across an await
   --> src/main.rs:224:27
    |
221 |     let mut stream = Post::find().stream(conn).await.unwrap();
    |         ---------- has type `impl futures::Stream<Item = Result<<post::Entity as sea_orm::EntityTrait>::Model, sea_orm::DbErr>>` which is not `Send`
...
224 |     while let Some(row) = stream.try_next().await.unwrap() {
    |                           ^^^^^^^^^^^^^^^^^^^^^^^ await occurs here, with `mut stream` maybe used later
...
229 | }
    | - `mut stream` is later dropped here
note: required by a bound in `__axum_debug_check_delete_post_future::check`
   --> src/main.rs:202:1
    |
202 | async fn delete_post(
    | ^^^^^ required by this bound in `__axum_debug_check_delete_post_future::check`

error: generator cannot be sent between threads safely
   --> src/main.rs:202:1
    |
202 | async fn delete_post(
    | ^^^^^ future returned by `delete_post` is not `Send`
    |
    = help: the trait `std::marker::Send` is not implemented for `dyn futures::Future<Output = Result<QueryStream, sea_orm::DbErr>>`
note: required by a bound in `__axum_debug_check_delete_post_future::check`
   --> src/main.rs:202:1
    |
202 | async fn delete_post(
    | ^^^^^ required by this bound in `__axum_debug_check_delete_post_future::check`

[sebpuetz]

More streamlined reproducer:

// post and setup are taken from axum_example unchanged
mod post;
mod setup;

use post::Entity as Post;
use sea_orm::{prelude::*, Database};
use std::env;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    env::set_var("RUST_LOG", "debug");
    tracing_subscriber::fmt::init();

    dotenv::dotenv().ok();
    let db_url = env::var("DATABASE_URL").expect("DATABASE_URL is not set in .env file");

    let conn = Database::connect(db_url)
        .await
        .expect("Database connection failed");
    tokio::spawn(async move { Post::find().stream(&conn).await });

    Ok(())
}

Error:

    Checking sea-orm-axum-example v0.1.0 (/home/sebastian/src/sea-orm/examples/axum_example)
error[E0277]: `dyn futures_core::stream::Stream<Item = Result<post::Model, sea_orm::DbErr>>` cannot be sent between threads safely
   --> src/main.rs:23:18
    |
23  |     tokio::spawn(async move { Post::find().stream(&conn).await });
    |     ------------ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ `dyn futures_core::stream::Stream<Item = Result<post::Model, sea_orm::DbErr>>` cannot be sent between threads safely
    |     |
    |     required by a bound introduced by this call
    |
    = help: the trait `Send` is not implemented for `dyn futures_core::stream::Stream<Item = Result<post::Model, sea_orm::DbErr>>`
    = note: required because of the requirements on the impl of `Send` for `Unique<dyn futures_core::stream::Stream<Item = Result<post::Model, sea_orm::DbErr>>>`
    = note: required because it appears within the type `std::boxed::Box<dyn futures_core::stream::Stream<Item = Result<post::Model, sea_orm::DbErr>>>`
    = note: required because it appears within the type `Pin<std::boxed::Box<dyn futures_core::stream::Stream<Item = Result<post::Model, sea_orm::DbErr>>>>`
    = note: required because it appears within the type `impl futures_core::stream::Stream<Item = Result<<post::Entity as sea_orm::EntityTrait>::Model, sea_orm::DbErr>>`
    = note: required because it appears within the type `Result<impl futures_core::stream::Stream<Item = Result<<post::Entity as sea_orm::EntityTrait>::Model, sea_orm::DbErr>>, sea_orm::DbErr>`
note: required by a bound in `tokio::spawn`
   --> /home/sebastian/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-1.15.0/src/task/spawn.rs:128:20
    |
128 |         T::Output: Send + 'static,
    |                    ^^^^ required by this bound in `tokio::spawn`

For more information about this error, try `rustc --explain E0277`.
error: could not compile `sea-orm-axum-example` due to previous error

stream() is !Send but tokio::spawn requires Future<...> + Send, thus this fails. It's possible to work around this limitation with tokio::task::LocalSet but it's way more involved than what you'd usually like because LocalSets can't be spawned within a running tokio::task so there's a fairly long tail of changes following it.

[billy1624]

Hey @sebpuetz, thanks for the explanation