Something is not Sync about AsyncStream (PgNotifications)

[crocme10]

Hello,

I am trying to publish Postgres notifications using Server Sent Events (SSE). I have sqlx to connect to Postgres, and warp to server SSEs.

The code does not compile, complaining that something is not Sync, and breaks the requirements on the function reply

dyn core::future::future::Future<Output = std::result::Result<(), sqlx_core::error::Error>> + std::marker::Send` cannot be shared between threads safely

I am not sure where this problem originates, as sqlx::Error is Sync, PgNotification is Sync....

Thank you for your help

[crocme10]

Hello,

this issue hasn't been met with much interest so far, and it is still something I'd like to figure out. I wonder how I can make it easier to reproduce, or more palatable....

The things to note are:

1. It's at the boundary between two packages (sqlx and warp) so it does not help. That being said, warp specifies clearly its constraints, and I thought that the notification stream would meet these requirements.
2. I cannot include error messages, because I've never seen in my rust experience bigger error messages than the one you get compiling this example. It reminds me of errors produced by gcc when it deals with template!

Anyway, I include the two pieces of code in the link, because this is pretty short, and because I hope someone will give me a hint, or a no go!

Cargo.toml:

[package]
name = "test-sse"
version = "0.1.0"
authors = ["riendegris"]
edition = "2018"

[dependencies]
warp = "0.2.3"
tokio = { version = "0.2", features = [ "macros" ] }
futures = "0.3"
sqlx = { version = "0.3.4", default-features = false, features = [ "postgres", "runtime-tokio", "macros" ] }

use futures::{Stream, StreamExt};
use sqlx::postgres::PgListener;
use std::convert::Infallible;
use warp::{
    sse::{self, ServerSentEvent},
    Filter,
};

#[tokio::main]
async fn main() -> Result<(), Infallible> {
    let notifications = warp::path("notifications")
        .and(warp::get())
        .and_then(|| async move {
            let stream: Result<_, Infallible> = Ok(warp::sse::reply(make_stream().await));
            stream
        });

    let index = warp::fs::file("index.html");

    let dir = warp::fs::dir("static");

    let routes = notifications.or(index).or(dir);

    warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;

    Ok(())
}

async fn make_stream() -> impl Stream<Item = Result<impl ServerSentEvent, Infallible>> {
    let mut listener = PgListener::new("DATABASE_URL").await.unwrap();
    listener.listen("channel").await.unwrap();
    let stream = listener.into_stream();
    stream.map(|i| {
        let n = i.unwrap();
        Ok(sse::event(String::from(n.channel())))
    })
}

[arlyon]

I've had similar troubles trying to get the listener working with tide SSE. Here's some example code which at least rules out actix from being the issue.

use std::convert::TryFrom;
use std::env;
use std::str::FromStr;

use anyhow::Result;
use sqlx::postgres::{PgListener, PgPool};
use tide::{Body, http::mime, Request, Response, StatusCode, sse};
use uuid::Uuid;

stuct AppState {
    pool: PgPool
}

#[async_std::main]
async fn main() -> Result<()> {
    let listen_addr = env::var("LISTEN_ADDR").unwrap_or_else(|_| "localhost:8000".to_owned());

    let pool = database_connection().await?;
    let app_state = AppState { pool };
    let mut app = tide::with_state(app_state);

    app.at("/scale/:uuid").get(sse::endpoint(|req, sender| async move {
        let scale: Uuid = req.param("uuid")?;

        let pool = req.state().pool.clone();
        let mut listener = PgListener::from_pool(&pool.clone()).await.unwrap();
        listener.listen(&scale.to_string());

        loop {
            let notification = listener.recv().await.unwrap().payload();
            sender.send(&scale.to_string(), notification, None).await;
        }
    }));

    app.listen(listen_addr).await?;

    Ok(())
}

async fn database_connection() -> Result<PgPool> {
    ...
}

And (some) of the output:

error[E0277]: `dyn std::future::Future<Output = std::result::Result<(), sqlx_core::error::Error>> + std::marker::Send` cannot be shared between threads safely
  --> src/main.rs:63:32
   |
63 |     app.at("/scale/:uuid").get(sse::endpoint(|req, sender| async move {
   |                                ^^^^^^^^^^^^^ `dyn std::future::Future<Output = std::result::Result<(), sqlx_core::error::Error>> + std::marker::Send` cannot be shared between threads safely
   | 
  ::: /Users/arlyon/.cargo/registry/src/github.com-1ecc6299db9ec823/tide-0.11.0/src/sse/endpoint.rs:19:47
   |
19 |     Fut: Future<Output = Result<()>> + Send + Sync + 'static,
   |                                               ---- required by this bound in `tide::sse::endpoint::endpoint`
   |
   = help: the trait `std::marker::Sync` is not implemented for `dyn std::future::Future<Output = std::result::Result<(), sqlx_core::error::Error>> + std::marker::Send`
   = note: required because of the requirements on the impl of `std::marker::Sync` for `std::ptr::Unique<dyn std::future::Future<Output = std::result::Result<(), sqlx_core::error::Error>> + std::marker::Send>`
   = note: required because it appears within the type `std::boxed::Box<dyn std::future::Future<Output = std::result::Result<(), sqlx_core::error::Error>> + std::marker::Send>`
   = note: required because it appears within the type `std::pin::Pin<std::boxed::Box<dyn std::future::Future<Output = std::result::Result<(), sqlx_core::error::Error>> + std::marker::Send>>`
   = note: required because it appears within the type `for<'r, 's> {std::future::ResumeTy, sqlx_core::pool::connection::Floating<'r, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>, sqlx_core::pool::connection::Live<sqlx_core::postgres::database::Postgres>, sqlx_core::postgres::connection::PgConnection, std::pin::Pin<std::boxed::Box<(dyn std::future::Future<Output = std::result::Result<(), sqlx_core::error::Error>> + std::marker::Send + 's)>>, ()}`
   = note: required because it appears within the type `[static generator@sqlx_core::pool::connection::Floating::<'s, sqlx_core::pool::connection::Idle<DB>>::close::{{closure}}#0 0:sqlx_core::pool::connection::Floating<'_, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>> for<'r, 's> {std::future::ResumeTy, sqlx_core::pool::connection::Floating<'r, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>, sqlx_core::pool::connection::Live<sqlx_core::postgres::database::Postgres>, sqlx_core::postgres::connection::PgConnection, std::pin::Pin<std::boxed::Box<(dyn std::future::Future<Output = std::result::Result<(), sqlx_core::error::Error>> + std::marker::Send + 's)>>, ()}]`
   = note: required because it appears within the type `std::future::from_generator::GenFuture<[static generator@sqlx_core::pool::connection::Floating::<'s, sqlx_core::pool::connection::Idle<DB>>::close::{{closure}}#0 0:sqlx_core::pool::connection::Floating<'_, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>> for<'r, 's> {std::future::ResumeTy, sqlx_core::pool::connection::Floating<'r, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>, sqlx_core::pool::connection::Live<sqlx_core::postgres::database::Postgres>, sqlx_core::postgres::connection::PgConnection, std::pin::Pin<std::boxed::Box<(dyn std::future::Future<Output = std::result::Result<(), sqlx_core::error::Error>> + std::marker::Send + 's)>>, ()}]>`
   = note: required because it appears within the type `impl std::future::Future`
   = note: required because it appears within the type `impl std::future::Future`
   = note: required because it appears within the type `for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6> {std::future::ResumeTy, sqlx_core::pool::connection::Floating<'r, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, &'s sqlx_core::pool::options::Options, bool, impl std::future::Future, (), &'t2 mut sqlx_core::pool::connection::Floating<'t3, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, impl std::future::Future}`
   = note: required because it appears within the type `[static generator@sqlx_core::pool::inner::check_conn::{{closure}}#0 0:sqlx_core::pool::connection::Floating<'_, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, 1:&sqlx_core::pool::options::Options for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6> {std::future::ResumeTy, sqlx_core::pool::connection::Floating<'r, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, &'s sqlx_core::pool::options::Options, bool, impl std::future::Future, (), &'t2 mut sqlx_core::pool::connection::Floating<'t3, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, impl std::future::Future}]`
   = note: required because it appears within the type `std::future::from_generator::GenFuture<[static generator@sqlx_core::pool::inner::check_conn::{{closure}}#0 0:sqlx_core::pool::connection::Floating<'_, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, 1:&sqlx_core::pool::options::Options for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6> {std::future::ResumeTy, sqlx_core::pool::connection::Floating<'r, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, &'s sqlx_core::pool::options::Options, bool, impl std::future::Future, (), &'t2 mut sqlx_core::pool::connection::Floating<'t3, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, impl std::future::Future}]>`
   = note: required because it appears within the type `impl std::future::Future`
   = note: required because it appears within the type `impl std::future::Future`
   = note: required because it appears within the type `for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9> {std::future::ResumeTy, &'r sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>, std::time::Instant, bool, sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>, &'s crossbeam_queue::array_queue::ArrayQueue<sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, crossbeam_queue::array_queue::ArrayQueue<sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, std::result::Result<sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>, crossbeam_queue::err::PopError>, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>, sqlx_core::pool::connection::Floating<'t0, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, sqlx_core::pool::options::Options, &'t1 sqlx_core::pool::options::Options, impl std::future::Future, (), std::option::Option<sqlx_core::pool::inner::DecrementSizeGuard<'t6>>, sqlx_core::pool::inner::DecrementSizeGuard<'t7>, impl std::future::Future, impl std::future::Future}`
   = note: required because it appears within the type `[static generator@sqlx_core::pool::inner::SharedPool::<DB>::acquire::{{closure}}#0 0:&sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres> for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9> {std::future::ResumeTy, &'r sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>, std::time::Instant, bool, sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>, &'s crossbeam_queue::array_queue::ArrayQueue<sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, crossbeam_queue::array_queue::ArrayQueue<sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, std::result::Result<sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>, crossbeam_queue::err::PopError>, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>, sqlx_core::pool::connection::Floating<'t0, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, sqlx_core::pool::options::Options, &'t1 sqlx_core::pool::options::Options, impl std::future::Future, (), std::option::Option<sqlx_core::pool::inner::DecrementSizeGuard<'t6>>, sqlx_core::pool::inner::DecrementSizeGuard<'t7>, impl std::future::Future, impl std::future::Future}]`
   = note: required because it appears within the type `std::future::from_generator::GenFuture<[static generator@sqlx_core::pool::inner::SharedPool::<DB>::acquire::{{closure}}#0 0:&sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres> for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8, 't9> {std::future::ResumeTy, &'r sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>, std::time::Instant, bool, sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>, &'s crossbeam_queue::array_queue::ArrayQueue<sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, crossbeam_queue::array_queue::ArrayQueue<sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, std::result::Result<sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>, crossbeam_queue::err::PopError>, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>, sqlx_core::pool::connection::Floating<'t0, sqlx_core::pool::connection::Idle<sqlx_core::postgres::database::Postgres>>, sqlx_core::pool::options::Options, &'t1 sqlx_core::pool::options::Options, impl std::future::Future, (), std::option::Option<sqlx_core::pool::inner::DecrementSizeGuard<'t6>>, sqlx_core::pool::inner::DecrementSizeGuard<'t7>, impl std::future::Future, impl std::future::Future}]>`
   = note: required because it appears within the type `impl std::future::Future`
   = note: required because it appears within the type `impl std::future::Future`
   = note: required because it appears within the type `for<'r, 's> {std::future::ResumeTy, &'r sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>, std::sync::Arc<sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>>, impl std::future::Future, ()}`
   = note: required because it appears within the type `[static generator@sqlx_core::pool::Pool::<DB>::acquire::{{closure}}#0 0:std::sync::Arc<sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>> for<'r, 's> {std::future::ResumeTy, &'r sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>, std::sync::Arc<sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>>, impl std::future::Future, ()}]`
   = note: required because it appears within the type `std::future::from_generator::GenFuture<[static generator@sqlx_core::pool::Pool::<DB>::acquire::{{closure}}#0 0:std::sync::Arc<sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>> for<'r, 's> {std::future::ResumeTy, &'r sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>, std::sync::Arc<sqlx_core::pool::inner::SharedPool<sqlx_core::postgres::database::Postgres>>, impl std::future::Future, ()}]>`
   = note: required because it appears within the type `impl std::future::Future`
   = note: required because it appears within the type `impl std::future::Future`
   = note: required because it appears within the type `for<'r> {std::future::ResumeTy, &'r sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres>, impl std::future::Future, ()}`
   = note: required because it appears within the type `[static generator@sqlx_core::postgres::listener::PgListener::from_pool::{{closure}}#0 0:&sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres> for<'r> {std::future::ResumeTy, &'r sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres>, impl std::future::Future, ()}]`
   = note: required because it appears within the type `std::future::from_generator::GenFuture<[static generator@sqlx_core::postgres::listener::PgListener::from_pool::{{closure}}#0 0:&sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres> for<'r> {std::future::ResumeTy, &'r sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres>, impl std::future::Future, ()}]>`
   = note: required because it appears within the type `impl std::future::Future`
   = note: required because it appears within the type `impl std::future::Future`
   = note: required because it appears within the type `for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8> {std::future::ResumeTy, uuid::Uuid, &'r sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres>, sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres>, impl std::future::Future, (), sqlx_core::postgres::listener::PgListener, &'t0 tide::sse::sender::Sender, tide::sse::sender::Sender, &'t1 uuid::Uuid, std::string::String, &'t2 str, &'t3 std::string::String, std::option::Option<&'t4 str>, impl std::future::Future}`
   = note: required because it appears within the type `[static generator@src/main.rs:63:71: 73:6 req:tide::request::Request<AppState>, pool:sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres>, sender:tide::sse::sender::Sender for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8> {std::future::ResumeTy, uuid::Uuid, &'r sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres>, sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres>, impl std::future::Future, (), sqlx_core::postgres::listener::PgListener, &'t0 tide::sse::sender::Sender, tide::sse::sender::Sender, &'t1 uuid::Uuid, std::string::String, &'t2 str, &'t3 std::string::String, std::option::Option<&'t4 str>, impl std::future::Future}]`
   = note: required because it appears within the type `std::future::from_generator::GenFuture<[static generator@src/main.rs:63:71: 73:6 req:tide::request::Request<AppState>, pool:sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres>, sender:tide::sse::sender::Sender for<'r, 's, 't0, 't1, 't2, 't3, 't4, 't5, 't6, 't7, 't8> {std::future::ResumeTy, uuid::Uuid, &'r sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres>, sqlx_core::pool::Pool<sqlx_core::postgres::database::Postgres>, impl std::future::Future, (), sqlx_core::postgres::listener::PgListener, &'t0 tide::sse::sender::Sender, tide::sse::sender::Sender, &'t1 uuid::Uuid, std::string::String, &'t2 str, &'t3 std::string::String, std::option::Option<&'t4 str>, impl std::future::Future}]>`
   = note: required because it appears within the type `impl std::future::Future`

I've done some digging and haven't found any culrpits yet though admittedly I'm not sure how to best search for things that aren't send + sync

[arlyon]

Update:

I've tracked it down to some usages of BoxFuture: https://docs.rs/futures-core/0.3.5/futures_core/future/type.BoxFuture.html

Seems that anything that uses a BoxFuture (such as everything here: https://github.com/launchbadge/sqlx/blob/master/sqlx-core/src/connection.rs#L22) will throw errors...

This is where my knowledge of async spreads too thin to be of much use. Not sure of the implications, but I'm confident that the error I ran into from the lack of Sync on Result<(), Error>> is due to the fact that the ping function, that is used when getting a connection from the pool, isn't Sync. I'd like to guess that this is a consequence of rust not having async traits yet but I'm really not sure.

[mehcode]

Connections are not Sync which means PgListener needs to stay on one thread when its in-use. You could spawn (async spawn like from async-std or tokio) the listener loop and use a channel to communicate back to make it Sync.

I'm not sure the right answer to fix this situation. I haven't had time to dig into it but hopefully this information helps.