Outbound TCP Sockets

[kflansburg]

• Box JS types so that Socket is Sync / Send.
• Test Secure Transport On
• Test Allow Half Open
• Remove unwraps
• Test StartTls
• Test / clean up methods / attributes other than read and write

[kflansburg]

@zebp do you know what is going on with the build here? cargo test compiles fine locally.

[KianNH]

I suspect it's because of the target that workers-rs requires (wasm32-unknown-unknown), I get the same failures if I do cargo test --target wasm32-unknown-unknown.

I see historic issues (2017) that Mio doesn't support being compiled to WASM, unsure if it has changed since then.

[KianNH]

I couldn't really find any reasons why Mio would be compiled with all of the features disabled, and it turns out that since this project is using edition = "2018" that it's just a bug with the old resolver (https://doc.rust-lang.org/nightly/cargo/reference/resolver.html#resolver-versions).

Adding resolver = "2" to the Cargo.toml's [workspace] seems to have fixed it?

[workspace]
members = [
    "worker",
    "worker-build",
    "worker-macros",
    "worker-sandbox",
    "worker-sys",
]
+ resolver = "2"

[kflansburg]

Thanks, I'll try to dig into what is happening here, I've built a worker with this change, so somehow targeting Wasm in that context is not an issue.

[dom96]

Really cool! Left a few comments. Thank you for picking this up.

[KianNH]

🎉

If we can keep the resolver change in this PR then I can look at moving to the 2021 edition in a follow-up PR separately

[kflansburg]

Some notes from testing:

• sqlx requires some TLS implementation (I guess it manages it itself), so that may need some further investigation.
• tokio-postgres requires rand, not sure why, but that will need to be bypassed.
• Socket is not Send + Sync (even when using Box). This seems to be a weird quirk of wasm-bindgen-futures / JsValue which prevents Socket from being used with hyper.

Thoughts on the third one @zebp? Have you run into this before?

[TonyGiorgio]

Just wanted to tACK a2b09e9 - I'm testing this out in a project moving from using axum to cloudflare workers doing a websocket to tcp socket proxy and this is working well locally. The APIs are incredibly similar so it made copy-pasting work pretty well by just swapping out a few libs.

[zebp]

• Socket is not Send + Sync (even when using Box). This seems to be a weird quirk of wasm-bindgen-futures / JsValue which prevents Socket from being used with hyper.

We run into this quite a bit. Because the runtime only ever has one thread of execution for the Worker and we don't use SharedArrayBuffer we just do an unsafe impl Send/Sync.

[kflansburg]

I do have this working with hyper:

Cargo.toml

hyper = { version="0.14", features=['http1', 'client'], default-fetaures=false }
tokio = {version = "1.0", default-features=false, features=['io-util', 'macros']}

main.rs

async fn make_request(
    mut sender: hyper::client::conn::SendRequest<hyper::Body>,
    request: hyper::Request<hyper::Body>,
) -> Response {
    let hyper_response = sender.send_request(request).await.expect("send_request");
    let buf = hyper::body::to_bytes(hyper_response)
        .await
        .expect("collect body");
    let text = std::str::from_utf8(&buf).expect("utf-8");
    let mut response = Response::ok(text).unwrap();
    response
        .headers_mut()
        .append("Content-Type", "text/html")
        .ok();
    response
}

#[event(fetch)]
async fn main(_req: Request, _env: Env, _ctx: Context) -> Result<Response> {
    set_panic_hook();
    console_log!("Connecting...");
    let socket = Socket::builder().connect("example.com", 80)?;
    console_log!("Got socket!");

    let (request_sender, connection) = hyper::client::conn::handshake(socket)
        .await
        .expect("handshake");

    let request = hyper::Request::builder()
        // We need to manually add the host header because SendRequest does not
        .header("Host", "example.com")
        .method("GET")
        .body(hyper::Body::from(""))
        .expect("build request");

    tokio::select!(
        res = connection => {
            console_error!("Connection exited: {:?}", res);
            Err(worker::Error::RustError("Connection exited".to_string()))
        },
        response = make_request(request_sender, request) => {
            Ok(response)
        }
    )
}

[kflansburg]

Ok @zebp @dom96 I think this is ready for final review. Theres probably some room for refactoring for readability, but I think that can be done in a followup.

[kflansburg]

Full hyper example: https://github.com/kflansburg/hyper-on-workers

[kflansburg]

I think I'm pretty close to having hyper::Client working, but unfortunately it appears to rely on Instant::now, which I don't think we can patch. v1.0 might have some plumbing for configuring a custom timer, so this might have to wait until that stabilizes.

https://github.com/kflansburg/hyper-on-workers/blob/client/src/lib.rs

[zebp]

Mostly looks good, some nits in terms of style but we also need to add some testing via worker-sandbox. This will probably include a test that just spawns a tcp listener that'll expect a connection from the worker, which should be simple but it might involve us upgrading to miniflare 3 for the janky test suite.

[kflansburg]

Manual test for now located here: https://github.com/kflansburg/workers-rs-sockets-test

@dom96 @zebp

[zebp]

Still need to do another review pass but I played around with this to see if I could get tokio-postgres working and I was able to! The changes necessary to tokio-postgres are here if interested. This is really cool

use futures_util::FutureExt;
use tokio_postgres::{Config, NoTls};
use worker::*;

#[event(start)]
fn init() {
    console_error_panic_hook::set_once();
}

#[event(fetch)]
async fn main(req: Request, env: Env, _ctx: Context) -> Result<Response> {
    let router = Router::new().get_async("/", |_, ctx| async move {
        let socket = Socket::builder()
            .secure_transport(SecureTransport::Off)
            .connect("0.0.0.0", 5432)?;

        let (client, connection) = Config::new()
            .user("postgres")
            .password("postgres")
            .dbname("postgres")
            .connect_raw(socket, NoTls)
            .await
            .unwrap();

        let conn = connection.map(|r| r.unwrap());
        wasm_bindgen_futures::spawn_local(conn);

        let rows = client.query("select 10;", &[]).await.unwrap();
        let value: i32 = rows[0].get(0);

        Response::ok(format!("Hello from Rust! {}", value))
    });

    router.run(req, env).await
}

[kflansburg]

Awesome, I’m really excited about this!