Bad diagnostic for Send bound on Future coming from async fn in foreign crate

[estebank]

As documented in Debugging async generator errors, when the problematic generator is in a foreign crate, we don't give the nice errors for things being kept through an await point. The reason for that is this early return. We need to figure out a way of performing the subsequent analysis on foreign crates.

You can find some of my analysis so far in this thread.

CC @tmandry @davidtwco

error[E0277]: `(dyn futures::Stream<Item = std::result::Result<std::vec::Vec<repr::Row>, comm::Error>> + std::marker::Send + std::marker::Unpin + 'static)` cannot be shared between threads safely
   --> src/materialized/src/mux.rs:138:100
    |
138 |       async fn handle_connection(&self, conn: SniffedStream<TcpStream>) -> Result<(), anyhow::Error> {
    |  ____________________________________________________________________________________________________^
139 | |         self.handle_connection(conn).await
140 | |     }
    | |_____^ `(dyn futures::Stream<Item = std::result::Result<std::vec::Vec<repr::Row>, comm::Error>> + std::marker::Send + std::marker::Unpin + 'static)` cannot be shared between threads safely
    |
    = help: the trait `std::marker::Sync` is not implemented for `(dyn futures::Stream<Item = std::result::Result<std::vec::Vec<repr::Row>, comm::Error>> + std::marker::Send + std::marker::Unpin + 'static)`
    = note: required because of the requirements on the impl of `std::marker::Sync` for `std::ptr::Unique<(dyn futures::Stream<Item = std::result::Result<std::vec::Vec<repr::Row>, comm::Error>> + std::marker::Send + std::marker::Unpin + 'static)>`
    = note: required because it appears within the type `std::boxed::Box<(dyn futures::Stream<Item = std::result::Result<std::vec::Vec<repr::Row>, comm::Error>> + std::marker::Send + std::marker::Unpin + 'static)>`
    = note: required because it appears within the type `std::option::Option<std::boxed::Box<(dyn futures::Stream<Item = std::result::Result<std::vec::Vec<repr::Row>, comm::Error>> + std::marker::Send + std::marker::Unpin + 'static)>>`
    = note: required because it appears within the type `coord::session::Portal`
    = note: required because of the requirements on the impl of `std::marker::Send` for `&coord::session::Portal`
    = note: required because it appears within the type `for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8> {std::future::ResumeTy, &'r mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, std::string::String, pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, &'s mut coord::SessionClient, coord::SessionClient, &'t0 coord::session::Session, &'t1 mut coord::session::Session, &'t2 str, &'t3 std::string::String, std::option::Option<&'t4 coord::session::Portal>, tokio_postgres::error::sqlstate::SqlState, impl futures::Future, (), &'t7 coord::session::Portal, impl futures::Future}`
    = note: required because it appears within the type `[static generator@pgwire::protocol::StateMachine::<A>::describe_portal::#0 0:&mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, 1:std::string::String for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8> {std::future::ResumeTy, &'r mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, std::string::String, pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, &'s mut coord::SessionClient, coord::SessionClient, &'t0 coord::session::Session, &'t1 mut coord::session::Session, &'t2 str, &'t3 std::string::String, std::option::Option<&'t4 coord::session::Portal>, tokio_postgres::error::sqlstate::SqlState, impl futures::Future, (), &'t7 coord::session::Portal, impl futures::Future}]`
    = note: required because it appears within the type `std::future::from_generator::GenFuture<[static generator@pgwire::protocol::StateMachine::<A>::describe_portal::#0 0:&mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, 1:std::string::String for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8> {std::future::ResumeTy, &'r mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, std::string::String, pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, &'s mut coord::SessionClient, coord::SessionClient, &'t0 coord::session::Session, &'t1 mut coord::session::Session, &'t2 str, &'t3 std::string::String, std::option::Option<&'t4 coord::session::Portal>, tokio_postgres::error::sqlstate::SqlState, impl futures::Future, (), &'t7 coord::session::Portal, impl futures::Future}]>`
    = note: required because it appears within the type `impl futures::Future`
    = note: required because it appears within the type `impl futures::Future`
    = note: required because it appears within the type `for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10, 't11> {std::future::ResumeTy, &'r mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, &'s mut pgwire::codec::FramedConn<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, pgwire::codec::FramedConn<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, impl futures::Future, (), std::option::Option<pgwire::message::FrontendMessage>, std::time::Instant, &'t1 str, std::string::String, impl futures::Future, std::vec::Vec<u32>, impl futures::Future, std::vec::Vec<pgrepr::format::Format>, std::vec::Vec<std::option::Option<std::vec::Vec<u8>>>, impl futures::Future, i32, usize, impl futures::Future, impl futures::Future, impl futures::Future, impl futures::Future, impl futures::Future, impl futures::Future, impl futures::Future}`
    = note: required because it appears within the type `[static generator@pgwire::protocol::StateMachine::<A>::advance_ready::#0 0:&mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>> for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10, 't11> {std::future::ResumeTy, &'r mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, &'s mut pgwire::codec::FramedConn<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, pgwire::codec::FramedConn<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, impl futures::Future, (), std::option::Option<pgwire::message::FrontendMessage>, std::time::Instant, &'t1 str, std::string::String, impl futures::Future, std::vec::Vec<u32>, impl futures::Future, std::vec::Vec<pgrepr::format::Format>, std::vec::Vec<std::option::Option<std::vec::Vec<u8>>>, impl futures::Future, i32, usize, impl futures::Future, impl futures::Future, impl futures::Future, impl futures::Future, impl futures::Future, impl futures::Future, impl futures::Future}]`
    = note: required because it appears within the type `std::future::from_generator::GenFuture<[static generator@pgwire::protocol::StateMachine::<A>::advance_ready::#0 0:&mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>> for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10, 't11> {std::future::ResumeTy, &'r mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, &'s mut pgwire::codec::FramedConn<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, pgwire::codec::FramedConn<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, impl futures::Future, (), std::option::Option<pgwire::message::FrontendMessage>, std::time::Instant, &'t1 str, std::string::String, impl futures::Future, std::vec::Vec<u32>, impl futures::Future, std::vec::Vec<pgrepr::format::Format>, std::vec::Vec<std::option::Option<std::vec::Vec<u8>>>, impl futures::Future, i32, usize, impl futures::Future, impl futures::Future, impl futures::Future, impl futures::Future, impl futures::Future, impl futures::Future, impl futures::Future}]>`
    = note: required because it appears within the type `impl futures::Future`
    = note: required because it appears within the type `impl futures::Future`
    = note: required because it appears within the type `for<'r, 's, 't0, 't1> {std::future::ResumeTy, pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, i32, std::vec::Vec<(std::string::String, std::string::String)>, &'r mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, impl futures::Future, (), pgwire::protocol::State, impl futures::Future, impl futures::Future, coord::SessionClient, impl futures::Future}`
    = note: required because it appears within the type `[static generator@pgwire::protocol::StateMachine::<A>::run::#0 0:pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, 1:i32, 2:std::vec::Vec<(std::string::String, std::string::String)> for<'r, 's, 't0, 't1> {std::future::ResumeTy, pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, i32, std::vec::Vec<(std::string::String, std::string::String)>, &'r mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, impl futures::Future, (), pgwire::protocol::State, impl futures::Future, impl futures::Future, coord::SessionClient, impl futures::Future}]`
    = note: required because it appears within the type `std::future::from_generator::GenFuture<[static generator@pgwire::protocol::StateMachine::<A>::run::#0 0:pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, 1:i32, 2:std::vec::Vec<(std::string::String, std::string::String)> for<'r, 's, 't0, 't1> {std::future::ResumeTy, pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, i32, std::vec::Vec<(std::string::String, std::string::String)>, &'r mut pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, impl futures::Future, (), pgwire::protocol::State, impl futures::Future, impl futures::Future, coord::SessionClient, impl futures::Future}]>`
    = note: required because it appears within the type `impl futures::Future`
    = note: required because it appears within the type `impl futures::Future`
    = note: required because it appears within the type `for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10> {std::future::ResumeTy, &'r pgwire::Server, ore::netio::SniffedStream<tokio::net::TcpStream>, pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>, &'s mut pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>, impl futures::Future, (), std::result::Result<pgwire::message::FrontendStartupMessage, std::io::Error>, pgwire::message::FrontendStartupMessage, i32, std::vec::Vec<(std::string::String, std::string::String)>, u32, coord::SessionClient, pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, impl futures::Future, bool, pgwire::Server, &'t1 coord::Client, coord::Client, &'t2 mut coord::Client, impl futures::Future, &'t4 mut ore::netio::SniffedStream<tokio::net::TcpStream>, u8, [u8; 1], &'t5 [u8], &'t6 [u8; 1], tokio::io::util::write_all::WriteAll<'t7, ore::netio::SniffedStream<tokio::net::TcpStream>>, &'t8 openssl::ssl::SslAcceptor, impl futures::Future, tokio::io::util::write_all::WriteAll<'t10, pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>}`
    = note: required because it appears within the type `[static generator@pgwire::Server::handle_connection::#0 0:&pgwire::Server, 1:ore::netio::SniffedStream<tokio::net::TcpStream> for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10> {std::future::ResumeTy, &'r pgwire::Server, ore::netio::SniffedStream<tokio::net::TcpStream>, pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>, &'s mut pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>, impl futures::Future, (), std::result::Result<pgwire::message::FrontendStartupMessage, std::io::Error>, pgwire::message::FrontendStartupMessage, i32, std::vec::Vec<(std::string::String, std::string::String)>, u32, coord::SessionClient, pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, impl futures::Future, bool, pgwire::Server, &'t1 coord::Client, coord::Client, &'t2 mut coord::Client, impl futures::Future, &'t4 mut ore::netio::SniffedStream<tokio::net::TcpStream>, u8, [u8; 1], &'t5 [u8], &'t6 [u8; 1], tokio::io::util::write_all::WriteAll<'t7, ore::netio::SniffedStream<tokio::net::TcpStream>>, &'t8 openssl::ssl::SslAcceptor, impl futures::Future, tokio::io::util::write_all::WriteAll<'t10, pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>}]`
    = note: required because it appears within the type `std::future::from_generator::GenFuture<[static generator@pgwire::Server::handle_connection::#0 0:&pgwire::Server, 1:ore::netio::SniffedStream<tokio::net::TcpStream> for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9, 't10> {std::future::ResumeTy, &'r pgwire::Server, ore::netio::SniffedStream<tokio::net::TcpStream>, pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>, &'s mut pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>, impl futures::Future, (), std::result::Result<pgwire::message::FrontendStartupMessage, std::io::Error>, pgwire::message::FrontendStartupMessage, i32, std::vec::Vec<(std::string::String, std::string::String)>, u32, coord::SessionClient, pgwire::protocol::StateMachine<pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>, impl futures::Future, bool, pgwire::Server, &'t1 coord::Client, coord::Client, &'t2 mut coord::Client, impl futures::Future, &'t4 mut ore::netio::SniffedStream<tokio::net::TcpStream>, u8, [u8; 1], &'t5 [u8], &'t6 [u8; 1], tokio::io::util::write_all::WriteAll<'t7, ore::netio::SniffedStream<tokio::net::TcpStream>>, &'t8 openssl::ssl::SslAcceptor, impl futures::Future, tokio::io::util::write_all::WriteAll<'t10, pgwire::server::Conn<ore::netio::SniffedStream<tokio::net::TcpStream>>>}]>`
    = note: required because it appears within the type `impl futures::Future`
    = note: required because it appears within the type `impl futures::Future`
    = note: required because it appears within the type `for<'r, 's> {std::future::ResumeTy, &'r pgwire::Server, ore::netio::SniffedStream<tokio::net::TcpStream>, impl futures::Future, ()}`
    = note: required because it appears within the type `[static generator@src/materialized/src/mux.rs:138:100: 140:6 _self:&pgwire::Server, conn:ore::netio::SniffedStream<tokio::net::TcpStream> for<'r, 's> {std::future::ResumeTy, &'r pgwire::Server, ore::netio::SniffedStream<tokio::net::TcpStream>, impl futures::Future, ()}]`
    = note: required because it appears within the type `std::future::from_generator::GenFuture<[static generator@src/materialized/src/mux.rs:138:100: 140:6 _self:&pgwire::Server, conn:ore::netio::SniffedStream<tokio::net::TcpStream> for<'r, 's> {std::future::ResumeTy, &'r pgwire::Server, ore::netio::SniffedStream<tokio::net::TcpStream>, impl futures::Future, ()}]>`
    = note: required because it appears within the type `impl futures::Future`
    = note: required because it appears within the type `impl futures::Future`
    = note: required for the cast to the object type `dyn futures::Future<Output = std::result::Result<(), anyhow::Error>> + std::marker::Send`

A (non-minimal) reproduction branch is available at https://github.com/mjibson/materialize/tree/async-error-message.

[lqd]

Some more information:

• in the reproduction repo, the error appears in the materialized crate
• but as Esteban mentioned, the bad diagnostics require a foreign crate: in this case, the pgwire crate

I've reduced the materialized crate to this playground but it requires the async_trait crate and of course pgwire and its dependencies. (I'll see if I can reduce this one as well)

[lqd]

Current status:

• removed the dependency on async_trait from materialized, leaving only pgwire
• removed all but the coord dependency in pgwire

Those two result in materialized reduced and pgwire reduced (but will probably be minimized further).

I feel/hope coord should be the final step here, as only the Session is now used in the reductions, and that it will remove the 500 dependencies into 2 or so. This should be easy and I'll look into it tomorrow.

[lqd]

Alrighty, we're finally down to a manageable size, 100 lines or so in 2 crates

• materialized only depending on pgwire: playground
• pgwire, no dependencies: playground

They could for sure be reduced further (if someone wants to have fun), but doing so also significantly reduces the futures' obligation chain and the problematic amount of diagnostics described in the blog post and this issue.

[davidtwco]

I looked into this briefly today, as per @estebank's Twitter thread, this line seems to the culprit:

rust/compiler/rustc_trait_selection/src/traits/error_reporting/suggestions.rs

Lines 1371 to 1374 in 5d5ff84

	// Do not ICE on closure typeck (#66868).
	if !generator_did.is_local() {
	return false;
	}

Removing that results in an ICE because we cannot get the typeck results for the other crate - I've got to assume that adding typeck results to the cross-crate metadata would be bad for perf, but it might less bad if we just added the generator_interior_types of the generator, expr_ty_adjusted of the await expression and node_type of the upvar to the cross-crate metadata. I don't know how straightforward that would be to implement (not too awful, I think).

[tmandry]

Triage: This does seem like an important fix to make and @davidtwco describes a pretty clear path to implementing it. Marking on deck for wg-async-foundations.

[estebank]

@rustbot claim