Repository: bytecodealliance/wasmtime
Pull request #8282 (2024-04-02 13:02:24)

Separate TCP implementation from the host bindings
Part of #7705 was moving the implementation of TCP (and UDP) out of the host bindings and into their own types. This PR starts this process (only for TCP) by doing a simple lift and shift of the code from the host bindings into the `TcpSocket` type. This should make next steps easier as the border between host bindings and implementation become much clearer with this change.

In the future, we may choose to make `TcpSocket` a trait object allowing users to swap in their own implementations, but that's left for a future change.

You may notice that some of `TcpSocket`'s methods return `io::Result` while most return `SocketResult`. It would be nice to move `TcpSocket` over completely to `io::Result` so that the implementation is completely free of any host bindings. However, there are some unclear/missing mappings between `io::Error` and `ErrorCode`. For example, when the socket is in an invalid state, we currently return `ErrorCode::InvalidState`. However, there is no `io::Error` or `Errno` that maps 1 to 1 to this `ErrorCode`. We'll need to figure out exactly how we want to do this mapping, but I'm leaving that for a future PR.


Patch:
diff --git a/.github/actions/install-rust/action.yml b/.github/actions/install-rust/action.yml
index 41d528608..84fdd5b68 100644
--- a/.github/actions/install-rust/action.yml
+++ b/.github/actions/install-rust/action.yml
@@ -1,15 +1,11 @@
 name: 'Install Rust toolchain'
-description: 'Install a rust toolchain and cache the crates index'
+description: 'Install a rust toolchain'
 
 inputs:
   toolchain:
     description: 'Default toolchan to install'
     required: false
     default: 'default'
-  lockfiles:
-    description: 'Path glob for Cargo.lock files to use as cache keys'
-    required: false
-    default: '**/Cargo.lock'
 
 runs:
   using: composite
@@ -55,12 +51,6 @@ runs:
         EOF
         fi
 
-        # Use a more efficient method for fetching the crates.io-index than
-        # the (currently) default git-based index.
-        cat >> "$GITHUB_ENV" <<EOF
-        CARGO_REGISTRIES_CRATES_IO_PROTOCOL=sparse
-        EOF
-
     - name: Require semicolons in WIT
       shell: bash
       run: echo WIT_REQUIRE_SEMICOLONS=1 >> "$GITHUB_ENV"
@@ -68,44 +58,3 @@ runs:
     - name: Install the WASI target
       shell: bash
       run: rustup target add wasm32-wasi wasm32-unknown-unknown
-
-    - name: Choose registry cache key
-      shell: bash
-      # Update the registry index cache at most once per day. actions/cache
-      # won't write changes back to the cache if the cache key already exists,
-      # so this means every job may have to re-download the index entries which
-      # are new since the last time the cache key changed. Changing the cache
-      # key relatively frequently keeps the amount of duplicated work down. But
-      # changing it too frequently means we might hit the 10GB quota too
-      # quickly, which would cause GitHub to evict other caches we still want.
-      run: echo CARGO_REGISTRY_CACHE_KEY=$(date +%Y%m%d) >> $GITHUB_ENV
-
-    - name: Cache Cargo registry index
-      uses: actions/cache@v4
-      with:
-        path: ~/.cargo/registry/index/
-        key: cargo-registry-${{ env.CARGO_REGISTRY_CACHE_KEY }}
-        # Any older registry-index cache is still valid. It's a git clone, so
-        # cargo only has to pull down the changes since the index was cached.
-        restore-keys: cargo-registry-
-
-    - name: Cache crate sources for dependencies
-      uses: actions/cache@v4
-      with:
-        path: |
-          ~/.cargo/registry/cache/
-          ~/.cargo/git/db/
-        key: cargo-crates-${{ inputs.lockfiles }}-${{ hashFiles(inputs.lockfiles) }}
-        # If Cargo.lock has changed, we probably will need to get the source
-        # code for some crates we don't already have. But any crates we have
-        # source cached for are still valid. The only problem is nothing
-        # removes old crate sources from the cache, so using `restore-keys`
-        # this way may use more of our GitHub cache quota than we'd like.
-        #
-        # Also, scope this cache by which Cargo.lock we're building from.
-        # Otherwise, whichever job writes the cache first will get its
-        # dependencies cached, and that cache will be used as the basis for the
-        # next job, even though odds are pretty good the cache is useless.
-        restore-keys: cargo-crates-${{ inputs.lockfiles }}-
-
-# TODO: on cache miss, after cargo has updated the registry index, run `git gc`
diff --git a/.github/workflows/triage.yml b/.github/workflows/triage.yml
index c7bde9931..70229754a 100644
--- a/.github/workflows/triage.yml
+++ b/.github/workflows/triage.yml
@@ -25,14 +25,17 @@ jobs:
     - uses: bytecodealliance/labeler@schedule-fork
       with:
         repo-token: "${{ secrets.GITHUB_TOKEN }}"
+      continue-on-error: true
 
     # @-mention people who are subscribed to a label when an issue/PR is given
     # that label.
     - uses: bytecodealliance/subscribe-to-label-action@v1
       with:
         repo-token: "${{ secrets.GITHUB_TOKEN }}"
+      continue-on-error: true
 
     # Leave pre-determined comments on issues/PRs that are given a certain label.
     - uses: bytecodealliance/label-messager-action@v1
       with:
         repo-token: "${{ secrets.GITHUB_TOKEN }}"
+      continue-on-error: true
diff --git a/crates/runtime/src/instance.rs b/crates/runtime/src/instance.rs
index cf81701fd..0faa14d7c 100644
--- a/crates/runtime/src/instance.rs
+++ b/crates/runtime/src/instance.rs
@@ -801,7 +801,12 @@ impl Instance {
         // disconnected from the lifetime of `self`.
         let module = self.module().clone();
 
-        let empty = TableSegmentElements::Functions(Box::new([]));
+        // NB: fall back to an expressions-based list of elements which doesn't
+        // have static type information (as opposed to `Functions`) since we
+        // don't know just yet what type the table has. The type will be be
+        // inferred in the next step within `table_init_segment`.
+        let empty = TableSegmentElements::Expressions(Box::new([]));
+
         let elements = match module.passive_elements_map.get(&elem_index) {
             Some(index) if !self.dropped_elements.contains(elem_index) => {
                 &module.passive_elements[*index]
diff --git a/crates/wasi/src/host/network.rs b/crates/wasi/src/host/network.rs
index be832d448..287a4b5b4 100644
--- a/crates/wasi/src/host/network.rs
+++ b/crates/wasi/src/host/network.rs
@@ -212,29 +212,34 @@ impl From<cap_net_ext::AddressFamily> for IpAddressFamily {
 }
 
 pub(crate) mod util {
+    use std::io;
     use std::net::{IpAddr, Ipv6Addr, SocketAddr};
     use std::time::Duration;
 
-    use crate::bindings::sockets::network::ErrorCode;
     use crate::network::SocketAddressFamily;
-    use crate::SocketResult;
     use cap_net_ext::{AddressFamily, Blocking, UdpSocketExt};
     use rustix::fd::{AsFd, OwnedFd};
     use rustix::io::Errno;
     use rustix::net::sockopt;
 
-    pub fn validate_unicast(addr: &SocketAddr) -> SocketResult<()> {
+    pub fn validate_unicast(addr: &SocketAddr) -> io::Result<()> {
         match to_canonical(&addr.ip()) {
             IpAddr::V4(ipv4) => {
                 if ipv4.is_multicast() || ipv4.is_broadcast() {
-                    Err(ErrorCode::InvalidArgument.into())
+                    Err(io::Error::new(
+                        io::ErrorKind::InvalidInput,
+                        "Both IPv4 broadcast and multicast addresses are not supported",
+                    ))
                 } else {
                     Ok(())
                 }
             }
             IpAddr::V6(ipv6) => {
                 if ipv6.is_multicast() {
-                    Err(ErrorCode::InvalidArgument.into())
+                    Err(io::Error::new(
+                        io::ErrorKind::InvalidInput,
+                        "IPv6 multicast addresses are not supported",
+                    ))
                 } else {
                     Ok(())
                 }
@@ -242,13 +247,19 @@ pub(crate) mod util {
         }
     }
 
-    pub fn validate_remote_address(addr: &SocketAddr) -> SocketResult<()> {
+    pub fn validate_remote_address(addr: &SocketAddr) -> io::Result<()> {
         if to_canonical(&addr.ip()).is_unspecified() {
-            return Err(ErrorCode::InvalidArgument.into());
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "Remote address may not be `0.0.0.0` or `::`",
+            ));
         }
 
         if addr.port() == 0 {
-            return Err(ErrorCode::InvalidArgument.into());
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "Remote port may not be 0",
+            ));
         }
 
         Ok(())
@@ -257,7 +268,7 @@ pub(crate) mod util {
     pub fn validate_address_family(
         addr: &SocketAddr,
         socket_family: &SocketAddressFamily,
-    ) -> SocketResult<()> {
+    ) -> io::Result<()> {
         match (socket_family, addr.ip()) {
             (SocketAddressFamily::Ipv4, IpAddr::V4(_)) => Ok(()),
             (SocketAddressFamily::Ipv6, IpAddr::V6(ipv6)) => {
@@ -266,14 +277,23 @@ pub(crate) mod util {
                     // since 2006, OS handling of them is inconsistent and our own
                     // validations don't take them into account either.
                     // Note that these are not the same as IPv4-*mapped* IPv6 addresses.
-                    Err(ErrorCode::InvalidArgument.into())
+                    Err(io::Error::new(
+                        io::ErrorKind::InvalidInput,
+                        "IPv4-compatible IPv6 addresses are not supported",
+                    ))
                 } else if ipv6.to_ipv4_mapped().is_some() {
-                    Err(ErrorCode::InvalidArgument.into())
+                    Err(io::Error::new(
+                        io::ErrorKind::InvalidInput,
+                        "IPv4-mapped IPv6 address passed to an IPv6-only socket",
+                    ))
                 } else {
                     Ok(())
                 }
             }
-            _ => Err(ErrorCode::InvalidArgument.into()),
+            _ => Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "Address family mismatch",
+            )),
         }
     }
 
@@ -301,7 +321,7 @@ pub(crate) mod util {
      * Syscalls wrappers with (opinionated) portability fixes.
      */
 
-    pub fn udp_socket(family: AddressFamily, blocking: Blocking) -> std::io::Result<OwnedFd> {
+    pub fn udp_socket(family: AddressFamily, blocking: Blocking) -> io::Result<OwnedFd> {
         // Delegate socket creation to cap_net_ext. They handle a couple of things for us:
         // - On Windows: call WSAStartup if not done before.
         // - Set the NONBLOCK and CLOEXEC flags. Either immediately during socket creation,
diff --git a/crates/wasi/src/host/tcp.rs b/crates/wasi/src/host/tcp.rs
index 11d44d882..d0ccaaa42 100644
--- a/crates/wasi/src/host/tcp.rs
+++ b/crates/wasi/src/host/tcp.rs
@@ -1,22 +1,14 @@
-use crate::host::network::util;
 use crate::network::SocketAddrUse;
-use crate::runtime::with_ambient_tokio_runtime;
-use crate::tcp::{TcpReadStream, TcpSocket, TcpState, TcpWriteStream};
 use crate::{
     bindings::{
         io::streams::{InputStream, OutputStream},
-        sockets::network::{ErrorCode, IpAddressFamily, IpSocketAddress, Network},
+        sockets::network::{IpAddressFamily, IpSocketAddress, Network},
         sockets::tcp::{self, ShutdownType},
     },
     network::SocketAddressFamily,
 };
 use crate::{Pollable, SocketResult, WasiView};
-use io_lifetimes::AsSocketlike;
-use rustix::io::Errno;
-use rustix::net::sockopt;
 use std::net::SocketAddr;
-use std::sync::Arc;
-use std::task::Poll;
 use std::time::Duration;
 use wasmtime::component::Resource;
 
@@ -31,60 +23,14 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
     ) -> SocketResult<()> {
         self.ctx().allowed_network_uses.check_allowed_tcp()?;
         let table = self.table();
-        let socket = table.get(&this)?;
         let network = table.get(&network)?;
         let local_address: SocketAddr = local_address.into();
 
-        let tokio_socket = match &socket.tcp_state {
-            TcpState::Default(socket) => socket,
-            TcpState::BindStarted(..) => return Err(ErrorCode::ConcurrencyConflict.into()),
-            _ => return Err(ErrorCode::InvalidState.into()),
-        };
-
-        util::validate_unicast(&local_address)?;
-        util::validate_address_family(&local_address, &socket.family)?;
-
-        {
-            // Ensure that we're allowed to connect to this address.
-            network.check_socket_addr(&local_address, SocketAddrUse::TcpBind)?;
-
-            // Automatically bypass the TIME_WAIT state when the user is trying
-            // to bind to a specific port:
-            let reuse_addr = local_address.port() > 0;
-
-            // Unconditionally (re)set SO_REUSEADDR, even when the value is false.
-            // This ensures we're not accidentally affected by any socket option
-            // state left behind by a previous failed call to this method (start_bind).
-            util::set_tcp_reuseaddr(&tokio_socket, reuse_addr)?;
-
-            // Perform the OS bind call.
-            tokio_socket.bind(local_address).map_err(|error| {
-                match Errno::from_io_error(&error) {
-                    // From https://pubs.opengroup.org/onlinepubs/9699919799/functions/bind.html:
-                    // > [EAFNOSUPPORT] The specified address is not a valid address for the address family of the specified socket
-                    //
-                    // The most common reasons for this error should have already
-                    // been handled by our own validation slightly higher up in this
-                    // function. This error mapping is here just in case there is
-                    // an edge case we didn't catch.
-                    Some(Errno::AFNOSUPPORT) => ErrorCode::InvalidArgument,
-
-                    // See: https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-bind#:~:text=WSAENOBUFS
-                    // Windows returns WSAENOBUFS when the ephemeral ports have been exhausted.
-                    #[cfg(windows)]
-                    Some(Errno::NOBUFS) => ErrorCode::AddressInUse,
-
-                    _ => ErrorCode::from(error),
-                }
-            })?;
-        }
-
-        let socket = table.get_mut(&this)?;
+        // Ensure that we're allowed to connect to this address.
+        network.check_socket_addr(&local_address, SocketAddrUse::TcpBind)?;
 
-        socket.tcp_state = match std::mem::replace(&mut socket.tcp_state, TcpState::Closed) {
-            TcpState::Default(socket) => TcpState::BindStarted(socket),
-            _ => unreachable!(),
-        };
+        // Bind to the address.
+        table.get_mut(&this)?.start_bind(local_address)?;
 
         Ok(())
     }
@@ -93,17 +39,7 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
         let table = self.table();
         let socket = table.get_mut(&this)?;
 
-        match socket.tcp_state {
-            TcpState::BindStarted(..) => {}
-            _ => return Err(ErrorCode::NotInProgress.into()),
-        }
-
-        socket.tcp_state = match std::mem::replace(&mut socket.tcp_state, TcpState::Closed) {
-            TcpState::BindStarted(socket) => TcpState::Bound(socket),
-            _ => unreachable!(),
-        };
-
-        Ok(())
+        socket.finish_bind()
     }
 
     fn start_connect(
@@ -114,37 +50,14 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
     ) -> SocketResult<()> {
         self.ctx().allowed_network_uses.check_allowed_tcp()?;
         let table = self.table();
-        let socket = table.get(&this)?;
         let network = table.get(&network)?;
         let remote_address: SocketAddr = remote_address.into();
 
-        match socket.tcp_state {
-            TcpState::Default(..) => {}
-
-            TcpState::Connecting(..) | TcpState::ConnectReady(..) => {
-                return Err(ErrorCode::ConcurrencyConflict.into())
-            }
-
-            _ => return Err(ErrorCode::InvalidState.into()),
-        };
-
-        util::validate_unicast(&remote_address)?;
-        util::validate_remote_address(&remote_address)?;
-        util::validate_address_family(&remote_address, &socket.family)?;
-
         // Ensure that we're allowed to connect to this address.
         network.check_socket_addr(&remote_address, SocketAddrUse::TcpConnect)?;
 
-        let socket = table.get_mut(&this)?;
-        let TcpState::Default(tokio_socket) =
-            std::mem::replace(&mut socket.tcp_state, TcpState::Closed)
-        else {
-            unreachable!();
-        };
-
-        let future = tokio_socket.connect(remote_address);
-
-        socket.tcp_state = TcpState::Connecting(Box::pin(future));
+        // Start connection
+        table.get_mut(&this)?.start_connect(remote_address)?;
 
         Ok(())
     }
@@ -156,45 +69,12 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
         let table = self.table();
         let socket = table.get_mut(&this)?;
 
-        let previous_state = std::mem::replace(&mut socket.tcp_state, TcpState::Closed);
-        let result = match previous_state {
-            TcpState::ConnectReady(result) => result,
-            TcpState::Connecting(mut future) => {
-                let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
-                match with_ambient_tokio_runtime(|| future.as_mut().poll(&mut cx)) {
-                    Poll::Ready(result) => result,
-                    Poll::Pending => {
-                        socket.tcp_state = TcpState::Connecting(future);
-                        return Err(ErrorCode::WouldBlock.into());
-                    }
-                }
-            }
-            previous_state => {
-                socket.tcp_state = previous_state;
-                return Err(ErrorCode::NotInProgress.into());
-            }
-        };
+        let (input, output) = socket.finish_connect()?;
 
-        match result {
-            Ok(stream) => {
-                let stream = Arc::new(stream);
-
-                let input: InputStream =
-                    InputStream::Host(Box::new(TcpReadStream::new(stream.clone())));
-                let output: OutputStream = Box::new(TcpWriteStream::new(stream.clone()));
-
-                let input_stream = self.table().push_child(input, &this)?;
-                let output_stream = self.table().push_child(output, &this)?;
-
-                let socket = self.table().get_mut(&this)?;
-                socket.tcp_state = TcpState::Connected(stream);
-                Ok((input_stream, output_stream))
-            }
-            Err(err) => {
-                socket.tcp_state = TcpState::Closed;
-                Err(err.into())
-            }
-        }
+        let input_stream = self.table().push_child(input, &this)?;
+        let output_stream = self.table().push_child(output, &this)?;
+
+        Ok((input_stream, output_stream))
     }
 
     fn start_listen(&mut self, this: Resource<tcp::TcpSocket>) -> SocketResult<()> {
@@ -202,62 +82,13 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
         let table = self.table();
         let socket = table.get_mut(&this)?;
 
-        match std::mem::replace(&mut socket.tcp_state, TcpState::Closed) {
-            TcpState::Bound(tokio_socket) => {
-                socket.tcp_state = TcpState::ListenStarted(tokio_socket);
-                Ok(())
-            }
-            TcpState::ListenStarted(tokio_socket) => {
-                socket.tcp_state = TcpState::ListenStarted(tokio_socket);
-                Err(ErrorCode::ConcurrencyConflict.into())
-            }
-            previous_state => {
-                socket.tcp_state = previous_state;
-                Err(ErrorCode::InvalidState.into())
-            }
-        }
+        socket.start_listen()
     }
 
     fn finish_listen(&mut self, this: Resource<tcp::TcpSocket>) -> SocketResult<()> {
         let table = self.table();
         let socket = table.get_mut(&this)?;
-
-        let tokio_socket = match std::mem::replace(&mut socket.tcp_state, TcpState::Closed) {
-            TcpState::ListenStarted(tokio_socket) => tokio_socket,
-            previous_state => {
-                socket.tcp_state = previous_state;
-                return Err(ErrorCode::NotInProgress.into());
-            }
-        };
-
-        match with_ambient_tokio_runtime(|| tokio_socket.listen(socket.listen_backlog_size)) {
-            Ok(listener) => {
-                socket.tcp_state = TcpState::Listening {
-                    listener,
-                    pending_accept: None,
-                };
-                Ok(())
-            }
-            Err(err) => {
-                socket.tcp_state = TcpState::Closed;
-
-                Err(match Errno::from_io_error(&err) {
-                    // See: https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-listen#:~:text=WSAEMFILE
-                    // According to the docs, `listen` can return EMFILE on Windows.
-                    // This is odd, because we're not trying to create a new socket
-                    // or file descriptor of any kind. So we rewrite it to less
-                    // surprising error code.
-                    //
-                    // At the time of writing, this behavior has never been experimentally
-                    // observed by any of the wasmtime authors, so we're relying fully
-                    // on Microsoft's documentation here.
-                    #[cfg(windows)]
-                    Some(Errno::MFILE) => Errno::NOBUFS.into(),
-
-                    _ => err.into(),
-                })
-            }
-        }
+        socket.finish_listen()
     }
 
     fn accept(
@@ -272,90 +103,7 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
         let table = self.table();
         let socket = table.get_mut(&this)?;
 
-        let TcpState::Listening {
-            listener,
-            pending_accept,
-        } = &mut socket.tcp_state
-        else {
-            return Err(ErrorCode::InvalidState.into());
-        };
-
-        let result = match pending_accept.take() {
-            Some(result) => result,
-            None => {
-                let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
-                match with_ambient_tokio_runtime(|| listener.poll_accept(&mut cx))
-                    .map_ok(|(stream, _)| stream)
-                {
-                    Poll::Ready(result) => result,
-                    Poll::Pending => Err(Errno::WOULDBLOCK.into()),
-                }
-            }
-        };
-
-        let client = result.map_err(|err| match Errno::from_io_error(&err) {
-            // From: https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-accept#:~:text=WSAEINPROGRESS
-            // > WSAEINPROGRESS: A blocking Windows Sockets 1.1 call is in progress,
-            // > or the service provider is still processing a callback function.
-            //
-            // wasi-sockets doesn't have an equivalent to the EINPROGRESS error,
-            // because in POSIX this error is only returned by a non-blocking
-            // `connect` and wasi-sockets has a different solution for that.
-            #[cfg(windows)]
-            Some(Errno::INPROGRESS) => Errno::INTR.into(),
-
-            // Normalize Linux' non-standard behavior.
-            //
-            // From https://man7.org/linux/man-pages/man2/accept.2.html:
-            // > Linux accept() passes already-pending network errors on the
-            // > new socket as an error code from accept(). This behavior
-            // > differs from other BSD socket implementations. (...)
-            #[cfg(target_os = "linux")]
-            Some(
-                Errno::CONNRESET
-                | Errno::NETRESET
-                | Errno::HOSTUNREACH
-                | Errno::HOSTDOWN
-                | Errno::NETDOWN
-                | Errno::NETUNREACH
-                | Errno::PROTO
-                | Errno::NOPROTOOPT
-                | Errno::NONET
-                | Errno::OPNOTSUPP,
-            ) => Errno::CONNABORTED.into(),
-
-            _ => err,
-        })?;
-
-        #[cfg(target_os = "macos")]
-        {
-            // Manually inherit socket options from listener. We only have to
-            // do this on platforms that don't already do this automatically
-            // and only if a specific value was explicitly set on the listener.
-
-            if let Some(size) = socket.receive_buffer_size {
-                _ = util::set_socket_recv_buffer_size(&client, size); // Ignore potential error.
-            }
-
-            if let Some(size) = socket.send_buffer_size {
-                _ = util::set_socket_send_buffer_size(&client, size); // Ignore potential error.
-            }
-
-            // For some reason, IP_TTL is inherited, but IPV6_UNICAST_HOPS isn't.
-            if let (SocketAddressFamily::Ipv6, Some(ttl)) = (socket.family, socket.hop_limit) {
-                _ = util::set_ipv6_unicast_hops(&client, ttl); // Ignore potential error.
-            }
-
-            if let Some(value) = socket.keep_alive_idle_time {
-                _ = util::set_tcp_keepidle(&client, value); // Ignore potential error.
-            }
-        }
-
-        let client = Arc::new(client);
-
-        let input: InputStream = InputStream::Host(Box::new(TcpReadStream::new(client.clone())));
-        let output: OutputStream = Box::new(TcpWriteStream::new(client.clone()));
-        let tcp_socket = TcpSocket::from_state(TcpState::Connected(client), socket.family)?;
+        let (tcp_socket, input, output) = socket.accept()?;
 
         let tcp_socket = self.table().push(tcp_socket)?;
         let input_stream = self.table().push_child(input, &tcp_socket)?;
@@ -368,38 +116,21 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
         let table = self.table();
         let socket = table.get(&this)?;
 
-        let view = match socket.tcp_state {
-            TcpState::Default(..) => return Err(ErrorCode::InvalidState.into()),
-            TcpState::BindStarted(..) => return Err(ErrorCode::ConcurrencyConflict.into()),
-            _ => socket.as_std_view()?,
-        };
-
-        Ok(view.local_addr()?.into())
+        socket.local_address().map(Into::into)
     }
 
     fn remote_address(&mut self, this: Resource<tcp::TcpSocket>) -> SocketResult<IpSocketAddress> {
         let table = self.table();
         let socket = table.get(&this)?;
 
-        let view = match socket.tcp_state {
-            TcpState::Connected(..) => socket.as_std_view()?,
-            TcpState::Connecting(..) | TcpState::ConnectReady(..) => {
-                return Err(ErrorCode::ConcurrencyConflict.into())
-            }
-            _ => return Err(ErrorCode::InvalidState.into()),
-        };
-
-        Ok(view.peer_addr()?.into())
+        socket.remote_address().map(Into::into)
     }
 
     fn is_listening(&mut self, this: Resource<tcp::TcpSocket>) -> Result<bool, anyhow::Error> {
         let table = self.table();
         let socket = table.get(&this)?;
 
-        match socket.tcp_state {
-            TcpState::Listening { .. } => Ok(true),
-            _ => Ok(false),
-        }
+        Ok(socket.is_listening())
     }
 
     fn address_family(
@@ -409,7 +140,7 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
         let table = self.table();
         let socket = table.get(&this)?;
 
-        match socket.family {
+        match socket.address_family() {
             SocketAddressFamily::Ipv4 => Ok(IpAddressFamily::Ipv4),
             SocketAddressFamily::Ipv6 => Ok(IpAddressFamily::Ipv6),
         }
@@ -420,49 +151,19 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
         this: Resource<tcp::TcpSocket>,
         value: u64,
     ) -> SocketResult<()> {
-        const MIN_BACKLOG: u32 = 1;
-        const MAX_BACKLOG: u32 = i32::MAX as u32; // OS'es will most likely limit it down even further.
-
         let table = self.table();
         let socket = table.get_mut(&this)?;
 
-        if value == 0 {
-            return Err(ErrorCode::InvalidArgument.into());
-        }
-
         // Silently clamp backlog size. This is OK for us to do, because operating systems do this too.
-        let value = value
-            .try_into()
-            .unwrap_or(u32::MAX)
-            .clamp(MIN_BACKLOG, MAX_BACKLOG);
-
-        match &socket.tcp_state {
-            TcpState::Default(..) | TcpState::Bound(..) => {
-                // Socket not listening yet. Stash value for first invocation to `listen`.
-                socket.listen_backlog_size = value;
-
-                Ok(())
-            }
-            TcpState::Listening { listener, .. } => {
-                // Try to update the backlog by calling `listen` again.
-                // Not all platforms support this. We'll only update our own value if the OS supports changing the backlog size after the fact.
-
-                rustix::net::listen(&listener, value.try_into().unwrap())
-                    .map_err(|_| ErrorCode::NotSupported)?;
-
-                socket.listen_backlog_size = value;
-
-                Ok(())
-            }
-            _ => Err(ErrorCode::InvalidState.into()),
-        }
+        let value = value.try_into().unwrap_or(u32::MAX);
+
+        socket.set_listen_backlog_size(value)
     }
 
     fn keep_alive_enabled(&mut self, this: Resource<tcp::TcpSocket>) -> SocketResult<bool> {
         let table = self.table();
         let socket = table.get(&this)?;
-        let view = &*socket.as_std_view()?;
-        Ok(sockopt::get_socket_keepalive(view)?)
+        socket.keep_alive_enabled()
     }
 
     fn set_keep_alive_enabled(
@@ -472,15 +173,13 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
     ) -> SocketResult<()> {
         let table = self.table();
         let socket = table.get(&this)?;
-        let view = &*socket.as_std_view()?;
-        Ok(sockopt::set_socket_keepalive(view, value)?)
+        socket.set_keep_alive_enabled(value)
     }
 
     fn keep_alive_idle_time(&mut self, this: Resource<tcp::TcpSocket>) -> SocketResult<u64> {
         let table = self.table();
         let socket = table.get(&this)?;
-        let view = &*socket.as_std_view()?;
-        Ok(sockopt::get_tcp_keepidle(view)?.as_nanos() as u64)
+        Ok(socket.keep_alive_idle_time()?.as_nanos() as u64)
     }
 
     fn set_keep_alive_idle_time(
@@ -491,25 +190,13 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
         let table = self.table();
         let socket = table.get_mut(&this)?;
         let duration = Duration::from_nanos(value);
-        {
-            let view = &*socket.as_std_view()?;
-
-            util::set_tcp_keepidle(view, duration)?;
-        }
-
-        #[cfg(target_os = "macos")]
-        {
-            socket.keep_alive_idle_time = Some(duration);
-        }
-
-        Ok(())
+        socket.set_keep_alive_idle_time(duration)
     }
 
     fn keep_alive_interval(&mut self, this: Resource<tcp::TcpSocket>) -> SocketResult<u64> {
         let table = self.table();
         let socket = table.get(&this)?;
-        let view = &*socket.as_std_view()?;
-        Ok(sockopt::get_tcp_keepintvl(view)?.as_nanos() as u64)
+        Ok(socket.keep_alive_interval()?.as_nanos() as u64)
     }
 
     fn set_keep_alive_interval(
@@ -519,15 +206,13 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
     ) -> SocketResult<()> {
         let table = self.table();
         let socket = table.get(&this)?;
-        let view = &*socket.as_std_view()?;
-        Ok(util::set_tcp_keepintvl(view, Duration::from_nanos(value))?)
+        socket.set_keep_alive_interval(Duration::from_nanos(value))
     }
 
     fn keep_alive_count(&mut self, this: Resource<tcp::TcpSocket>) -> SocketResult<u32> {
         let table = self.table();
         let socket = table.get(&this)?;
-        let view = &*socket.as_std_view()?;
-        Ok(sockopt::get_tcp_keepcnt(view)?)
+        socket.keep_alive_count()
     }
 
     fn set_keep_alive_count(
@@ -537,50 +222,26 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
     ) -> SocketResult<()> {
         let table = self.table();
         let socket = table.get(&this)?;
-        let view = &*socket.as_std_view()?;
-        Ok(util::set_tcp_keepcnt(view, value)?)
+        socket.set_keep_alive_count(value)
     }
 
     fn hop_limit(&mut self, this: Resource<tcp::TcpSocket>) -> SocketResult<u8> {
         let table = self.table();
         let socket = table.get(&this)?;
-        let view = &*socket.as_std_view()?;
-
-        let ttl = match socket.family {
-            SocketAddressFamily::Ipv4 => util::get_ip_ttl(view)?,
-            SocketAddressFamily::Ipv6 => util::get_ipv6_unicast_hops(view)?,
-        };
-
-        Ok(ttl)
+        socket.hop_limit()
     }
 
     fn set_hop_limit(&mut self, this: Resource<tcp::TcpSocket>, value: u8) -> SocketResult<()> {
         let table = self.table();
         let socket = table.get_mut(&this)?;
-        {
-            let view = &*socket.as_std_view()?;
-
-            match socket.family {
-                SocketAddressFamily::Ipv4 => util::set_ip_ttl(view, value)?,
-                SocketAddressFamily::Ipv6 => util::set_ipv6_unicast_hops(view, value)?,
-            }
-        }
-
-        #[cfg(target_os = "macos")]
-        {
-            socket.hop_limit = Some(value);
-        }
-
-        Ok(())
+        socket.set_hop_limit(value)
     }
 
     fn receive_buffer_size(&mut self, this: Resource<tcp::TcpSocket>) -> SocketResult<u64> {
         let table = self.table();
         let socket = table.get(&this)?;
-        let view = &*socket.as_std_view()?;
 
-        let value = util::get_socket_recv_buffer_size(view)?;
-        Ok(value as u64)
+        Ok(socket.receive_buffer_size()? as u64)
     }
 
     fn set_receive_buffer_size(
@@ -591,27 +252,14 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
         let table = self.table();
         let socket = table.get_mut(&this)?;
         let value = value.try_into().unwrap_or(usize::MAX);
-        {
-            let view = &*socket.as_std_view()?;
-
-            util::set_socket_recv_buffer_size(view, value)?;
-        }
-
-        #[cfg(target_os = "macos")]
-        {
-            socket.receive_buffer_size = Some(value);
-        }
-
-        Ok(())
+        socket.set_receive_buffer_size(value)
     }
 
     fn send_buffer_size(&mut self, this: Resource<tcp::TcpSocket>) -> SocketResult<u64> {
         let table = self.table();
         let socket = table.get(&this)?;
-        let view = &*socket.as_std_view()?;
 
-        let value = util::get_socket_send_buffer_size(view)?;
-        Ok(value as u64)
+        Ok(socket.send_buffer_size()? as u64)
     }
 
     fn set_send_buffer_size(
@@ -622,18 +270,7 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
         let table = self.table();
         let socket = table.get_mut(&this)?;
         let value = value.try_into().unwrap_or(usize::MAX);
-        {
-            let view = &*socket.as_std_view()?;
-
-            util::set_socket_send_buffer_size(view, value)?;
-        }
-
-        #[cfg(target_os = "macos")]
-        {
-            socket.send_buffer_size = Some(value);
-        }
-
-        Ok(())
+        socket.set_send_buffer_size(value)
     }
 
     fn subscribe(&mut self, this: Resource<tcp::TcpSocket>) -> anyhow::Result<Resource<Pollable>> {
@@ -648,21 +285,12 @@ impl<T: WasiView> crate::host::tcp::tcp::HostTcpSocket for T {
         let table = self.table();
         let socket = table.get(&this)?;
 
-        let stream = match &socket.tcp_state {
-            TcpState::Connected(stream) => stream,
-            _ => return Err(ErrorCode::InvalidState.into()),
-        };
-
         let how = match shutdown_type {
             ShutdownType::Receive => std::net::Shutdown::Read,
             ShutdownType::Send => std::net::Shutdown::Write,
             ShutdownType::Both => std::net::Shutdown::Both,
         };
-
-        stream
-            .as_socketlike_view::<std::net::TcpStream>()
-            .shutdown(how)?;
-        Ok(())
+        Ok(socket.shutdown(how)?)
     }
 
     fn drop(&mut self, this: Resource<tcp::TcpSocket>) -> Result<(), anyhow::Error> {
diff --git a/crates/wasi/src/tcp.rs b/crates/wasi/src/tcp.rs
index bc970c3bf..e385874d9 100644
--- a/crates/wasi/src/tcp.rs
+++ b/crates/wasi/src/tcp.rs
@@ -1,16 +1,24 @@
+use crate::bindings::sockets::tcp::ErrorCode;
+use crate::host::network;
 use crate::network::SocketAddressFamily;
 use crate::runtime::{with_ambient_tokio_runtime, AbortOnDropJoinHandle};
-use crate::{HostInputStream, HostOutputStream, SocketResult, StreamError, Subscribe};
+use crate::{
+    HostInputStream, HostOutputStream, InputStream, OutputStream, SocketResult, StreamError,
+    Subscribe,
+};
 use anyhow::{Error, Result};
 use cap_net_ext::AddressFamily;
 use futures::Future;
 use io_lifetimes::views::SocketlikeView;
 use io_lifetimes::AsSocketlike;
+use rustix::io::Errno;
 use rustix::net::sockopt;
 use std::io;
 use std::mem;
+use std::net::SocketAddr;
 use std::pin::Pin;
 use std::sync::Arc;
+use std::task::Poll;
 
 /// Value taken from rust std library.
 const DEFAULT_BACKLOG: u32 = 128;
@@ -19,7 +27,7 @@ const DEFAULT_BACKLOG: u32 = 128;
 ///
 /// This represents the various states a socket can be in during the
 /// activities of binding, listening, accepting, and connecting.
-pub(crate) enum TcpState {
+enum TcpState {
     /// The initial state for a newly-created socket.
     Default(tokio::net::TcpSocket),
 
@@ -51,30 +59,612 @@ pub(crate) enum TcpState {
     Closed,
 }
 
+impl std::fmt::Debug for TcpState {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Default(_) => f.debug_tuple("Default").finish(),
+            Self::BindStarted(_) => f.debug_tuple("BindStarted").finish(),
+            Self::Bound(_) => f.debug_tuple("Bound").finish(),
+            Self::ListenStarted(_) => f.debug_tuple("ListenStarted").finish(),
+            Self::Listening { pending_accept, .. } => f
+                .debug_struct("Listening")
+                .field("pending_accept", pending_accept)
+                .finish(),
+            Self::Connecting(_) => f.debug_tuple("Connecting").finish(),
+            Self::ConnectReady(_) => f.debug_tuple("ConnectReady").finish(),
+            Self::Connected(_) => f.debug_tuple("Connected").finish(),
+            Self::Closed => write!(f, "Closed"),
+        }
+    }
+}
+
 /// A host TCP socket, plus associated bookkeeping.
-///
-/// The inner state is wrapped in an Arc because the same underlying socket is
-/// used for implementing the stream types.
 pub struct TcpSocket {
     /// The current state in the bind/listen/accept/connect progression.
-    pub(crate) tcp_state: TcpState,
+    tcp_state: TcpState,
 
     /// The desired listen queue size.
-    pub(crate) listen_backlog_size: u32,
+    listen_backlog_size: u32,
 
-    pub(crate) family: SocketAddressFamily,
+    family: SocketAddressFamily,
 
     // The socket options below are not automatically inherited from the listener
     // on all platforms. So we keep track of which options have been explicitly
     // set and manually apply those values to newly accepted clients.
     #[cfg(target_os = "macos")]
-    pub(crate) receive_buffer_size: Option<usize>,
+    receive_buffer_size: Option<usize>,
     #[cfg(target_os = "macos")]
-    pub(crate) send_buffer_size: Option<usize>,
+    send_buffer_size: Option<usize>,
     #[cfg(target_os = "macos")]
-    pub(crate) hop_limit: Option<u8>,
+    hop_limit: Option<u8>,
     #[cfg(target_os = "macos")]
-    pub(crate) keep_alive_idle_time: Option<std::time::Duration>,
+    keep_alive_idle_time: Option<std::time::Duration>,
+}
+
+impl TcpSocket {
+    /// Create a new socket in the given family.
+    pub fn new(family: AddressFamily) -> io::Result<Self> {
+        with_ambient_tokio_runtime(|| {
+            let (socket, family) = match family {
+                AddressFamily::Ipv4 => {
+                    let socket = tokio::net::TcpSocket::new_v4()?;
+                    (socket, SocketAddressFamily::Ipv4)
+                }
+                AddressFamily::Ipv6 => {
+                    let socket = tokio::net::TcpSocket::new_v6()?;
+                    sockopt::set_ipv6_v6only(&socket, true)?;
+                    (socket, SocketAddressFamily::Ipv6)
+                }
+            };
+
+            Self::from_state(TcpState::Default(socket), family)
+        })
+    }
+
+    /// Create a `TcpSocket` from an existing socket.
+    fn from_state(state: TcpState, family: SocketAddressFamily) -> io::Result<Self> {
+        Ok(Self {
+            tcp_state: state,
+            listen_backlog_size: DEFAULT_BACKLOG,
+            family,
+            #[cfg(target_os = "macos")]
+            receive_buffer_size: None,
+            #[cfg(target_os = "macos")]
+            send_buffer_size: None,
+            #[cfg(target_os = "macos")]
+            hop_limit: None,
+            #[cfg(target_os = "macos")]
+            keep_alive_idle_time: None,
+        })
+    }
+
+    fn as_std_view(&self) -> SocketResult<SocketlikeView<'_, std::net::TcpStream>> {
+        use crate::bindings::sockets::network::ErrorCode;
+
+        match &self.tcp_state {
+            TcpState::Default(socket) | TcpState::Bound(socket) => {
+                Ok(socket.as_socketlike_view::<std::net::TcpStream>())
+            }
+            TcpState::Connected(stream) => Ok(stream.as_socketlike_view::<std::net::TcpStream>()),
+            TcpState::Listening { listener, .. } => {
+                Ok(listener.as_socketlike_view::<std::net::TcpStream>())
+            }
+
+            TcpState::BindStarted(..)
+            | TcpState::ListenStarted(..)
+            | TcpState::Connecting(..)
+            | TcpState::ConnectReady(..)
+            | TcpState::Closed => Err(ErrorCode::InvalidState.into()),
+        }
+    }
+}
+
+impl TcpSocket {
+    pub fn start_bind(&mut self, local_address: SocketAddr) -> io::Result<()> {
+        let tokio_socket = match &self.tcp_state {
+            TcpState::Default(socket) => socket,
+            TcpState::BindStarted(..) => return Err(Errno::ALREADY.into()),
+            _ => return Err(Errno::ISCONN.into()),
+        };
+
+        network::util::validate_unicast(&local_address)?;
+        network::util::validate_address_family(&local_address, &self.family)?;
+
+        {
+            // Automatically bypass the TIME_WAIT state when the user is trying
+            // to bind to a specific port:
+            let reuse_addr = local_address.port() > 0;
+
+            // Unconditionally (re)set SO_REUSEADDR, even when the value is false.
+            // This ensures we're not accidentally affected by any socket option
+            // state left behind by a previous failed call to this method (start_bind).
+            network::util::set_tcp_reuseaddr(&tokio_socket, reuse_addr)?;
+
+            // Perform the OS bind call.
+            tokio_socket.bind(local_address).map_err(|error| {
+                match Errno::from_io_error(&error) {
+                    // From https://pubs.opengroup.org/onlinepubs/9699919799/functions/bind.html:
+                    // > [EAFNOSUPPORT] The specified address is not a valid address for the address family of the specified socket
+                    //
+                    // The most common reasons for this error should have already
+                    // been handled by our own validation slightly higher up in this
+                    // function. This error mapping is here just in case there is
+                    // an edge case we didn't catch.
+                    Some(Errno::AFNOSUPPORT) =>  io::Error::new(
+                        io::ErrorKind::InvalidInput,
+                        "The specified address is not a valid address for the address family of the specified socket",
+                    ),
+
+                    // See: https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-bind#:~:text=WSAENOBUFS
+                    // Windows returns WSAENOBUFS when the ephemeral ports have been exhausted.
+                    #[cfg(windows)]
+                    Some(Errno::NOBUFS) => io::Error::new(io::ErrorKind::AddrInUse, "no more free local ports"),
+
+                    _ => error,
+                }
+            })?;
+
+            self.tcp_state = match std::mem::replace(&mut self.tcp_state, TcpState::Closed) {
+                TcpState::Default(socket) => TcpState::BindStarted(socket),
+                _ => unreachable!(),
+            };
+
+            Ok(())
+        }
+    }
+
+    pub fn finish_bind(&mut self) -> SocketResult<()> {
+        match std::mem::replace(&mut self.tcp_state, TcpState::Closed) {
+            TcpState::BindStarted(socket) => {
+                self.tcp_state = TcpState::Bound(socket);
+                Ok(())
+            }
+            current_state => {
+                // Reset the state so that the outside world doesn't see this socket as closed
+                self.tcp_state = current_state;
+                Err(ErrorCode::NotInProgress.into())
+            }
+        }
+    }
+
+    pub fn start_connect(&mut self, remote_address: SocketAddr) -> SocketResult<()> {
+        match self.tcp_state {
+            TcpState::Default(..) => {}
+
+            TcpState::Connecting(..) | TcpState::ConnectReady(..) => {
+                return Err(ErrorCode::ConcurrencyConflict.into())
+            }
+
+            _ => return Err(ErrorCode::InvalidState.into()),
+        };
+
+        network::util::validate_unicast(&remote_address)?;
+        network::util::validate_remote_address(&remote_address)?;
+        network::util::validate_address_family(&remote_address, &self.family)?;
+
+        let TcpState::Default(tokio_socket) =
+            std::mem::replace(&mut self.tcp_state, TcpState::Closed)
+        else {
+            unreachable!();
+        };
+
+        let future = tokio_socket.connect(remote_address);
+
+        self.tcp_state = TcpState::Connecting(Box::pin(future));
+        Ok(())
+    }
+
+    pub fn finish_connect(&mut self) -> SocketResult<(InputStream, OutputStream)> {
+        let previous_state = std::mem::replace(&mut self.tcp_state, TcpState::Closed);
+        let result = match previous_state {
+            TcpState::ConnectReady(result) => result,
+            TcpState::Connecting(mut future) => {
+                let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
+                match with_ambient_tokio_runtime(|| future.as_mut().poll(&mut cx)) {
+                    Poll::Ready(result) => result,
+                    Poll::Pending => {
+                        self.tcp_state = TcpState::Connecting(future);
+                        return Err(ErrorCode::WouldBlock.into());
+                    }
+                }
+            }
+            previous_state => {
+                self.tcp_state = previous_state;
+                return Err(ErrorCode::NotInProgress.into());
+            }
+        };
+
+        match result {
+            Ok(stream) => {
+                let stream = Arc::new(stream);
+                self.tcp_state = TcpState::Connected(stream.clone());
+                let input: InputStream =
+                    InputStream::Host(Box::new(TcpReadStream::new(stream.clone())));
+                let output: OutputStream = Box::new(TcpWriteStream::new(stream));
+                Ok((input, output))
+            }
+            Err(err) => {
+                self.tcp_state = TcpState::Closed;
+                Err(err.into())
+            }
+        }
+    }
+
+    pub fn start_listen(&mut self) -> SocketResult<()> {
+        match std::mem::replace(&mut self.tcp_state, TcpState::Closed) {
+            TcpState::Bound(tokio_socket) => {
+                self.tcp_state = TcpState::ListenStarted(tokio_socket);
+                Ok(())
+            }
+            TcpState::ListenStarted(tokio_socket) => {
+                self.tcp_state = TcpState::ListenStarted(tokio_socket);
+                Err(ErrorCode::ConcurrencyConflict.into())
+            }
+            previous_state => {
+                self.tcp_state = previous_state;
+                Err(ErrorCode::InvalidState.into())
+            }
+        }
+    }
+
+    pub fn finish_listen(&mut self) -> SocketResult<()> {
+        let tokio_socket = match std::mem::replace(&mut self.tcp_state, TcpState::Closed) {
+            TcpState::ListenStarted(tokio_socket) => tokio_socket,
+            previous_state => {
+                self.tcp_state = previous_state;
+                return Err(ErrorCode::NotInProgress.into());
+            }
+        };
+
+        match with_ambient_tokio_runtime(|| tokio_socket.listen(self.listen_backlog_size)) {
+            Ok(listener) => {
+                self.tcp_state = TcpState::Listening {
+                    listener,
+                    pending_accept: None,
+                };
+                Ok(())
+            }
+            Err(err) => {
+                self.tcp_state = TcpState::Closed;
+
+                Err(match Errno::from_io_error(&err) {
+                    // See: https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-listen#:~:text=WSAEMFILE
+                    // According to the docs, `listen` can return EMFILE on Windows.
+                    // This is odd, because we're not trying to create a new socket
+                    // or file descriptor of any kind. So we rewrite it to less
+                    // surprising error code.
+                    //
+                    // At the time of writing, this behavior has never been experimentally
+                    // observed by any of the wasmtime authors, so we're relying fully
+                    // on Microsoft's documentation here.
+                    #[cfg(windows)]
+                    Some(Errno::MFILE) => Errno::NOBUFS.into(),
+
+                    _ => err.into(),
+                })
+            }
+        }
+    }
+
+    pub fn accept(&mut self) -> SocketResult<(Self, InputStream, OutputStream)> {
+        let TcpState::Listening {
+            listener,
+            pending_accept,
+        } = &mut self.tcp_state
+        else {
+            return Err(ErrorCode::InvalidState.into());
+        };
+
+        let result = match pending_accept.take() {
+            Some(result) => result,
+            None => {
+                let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
+                match with_ambient_tokio_runtime(|| listener.poll_accept(&mut cx))
+                    .map_ok(|(stream, _)| stream)
+                {
+                    Poll::Ready(result) => result,
+                    Poll::Pending => Err(Errno::WOULDBLOCK.into()),
+                }
+            }
+        };
+
+        let client = result.map_err(|err| match Errno::from_io_error(&err) {
+            // From: https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-accept#:~:text=WSAEINPROGRESS
+            // > WSAEINPROGRESS: A blocking Windows Sockets 1.1 call is in progress,
+            // > or the service provider is still processing a callback function.
+            //
+            // wasi-sockets doesn't have an equivalent to the EINPROGRESS error,
+            // because in POSIX this error is only returned by a non-blocking
+            // `connect` and wasi-sockets has a different solution for that.
+            #[cfg(windows)]
+            Some(Errno::INPROGRESS) => Errno::INTR.into(),
+
+            // Normalize Linux' non-standard behavior.
+            //
+            // From https://man7.org/linux/man-pages/man2/accept.2.html:
+            // > Linux accept() passes already-pending network errors on the
+            // > new socket as an error code from accept(). This behavior
+            // > differs from other BSD socket implementations. (...)
+            #[cfg(target_os = "linux")]
+            Some(
+                Errno::CONNRESET
+                | Errno::NETRESET
+                | Errno::HOSTUNREACH
+                | Errno::HOSTDOWN
+                | Errno::NETDOWN
+                | Errno::NETUNREACH
+                | Errno::PROTO
+                | Errno::NOPROTOOPT
+                | Errno::NONET
+                | Errno::OPNOTSUPP,
+            ) => Errno::CONNABORTED.into(),
+
+            _ => err,
+        })?;
+
+        #[cfg(target_os = "macos")]
+        {
+            // Manually inherit socket options from listener. We only have to
+            // do this on platforms that don't already do this automatically
+            // and only if a specific value was explicitly set on the listener.
+
+            if let Some(size) = self.receive_buffer_size {
+                _ = network::util::set_socket_recv_buffer_size(&client, size); // Ignore potential error.
+            }
+
+            if let Some(size) = self.send_buffer_size {
+                _ = network::util::set_socket_send_buffer_size(&client, size); // Ignore potential error.
+            }
+
+            // For some reason, IP_TTL is inherited, but IPV6_UNICAST_HOPS isn't.
+            if let (SocketAddressFamily::Ipv6, Some(ttl)) = (self.family, self.hop_limit) {
+                _ = network::util::set_ipv6_unicast_hops(&client, ttl); // Ignore potential error.
+            }
+
+            if let Some(value) = self.keep_alive_idle_time {
+                _ = network::util::set_tcp_keepidle(&client, value); // Ignore potential error.
+            }
+        }
+
+        let client = Arc::new(client);
+
+        let input: InputStream = InputStream::Host(Box::new(TcpReadStream::new(client.clone())));
+        let output: OutputStream = Box::new(TcpWriteStream::new(client.clone()));
+        let tcp_socket = TcpSocket::from_state(TcpState::Connected(client), self.family)?;
+
+        Ok((tcp_socket, input, output))
+    }
+
+    pub fn local_address(&self) -> SocketResult<SocketAddr> {
+        let view = match self.tcp_state {
+            TcpState::Default(..) => return Err(ErrorCode::InvalidState.into()),
+            TcpState::BindStarted(..) => return Err(ErrorCode::ConcurrencyConflict.into()),
+            _ => self.as_std_view()?,
+        };
+
+        Ok(view.local_addr()?)
+    }
+
+    pub fn remote_address(&self) -> SocketResult<SocketAddr> {
+        let view = match self.tcp_state {
+            TcpState::Connected(..) => self.as_std_view()?,
+            TcpState::Connecting(..) | TcpState::ConnectReady(..) => {
+                return Err(ErrorCode::ConcurrencyConflict.into())
+            }
+            _ => return Err(ErrorCode::InvalidState.into()),
+        };
+
+        Ok(view.peer_addr()?)
+    }
+
+    pub fn is_listening(&self) -> bool {
+        matches!(self.tcp_state, TcpState::Listening { .. })
+    }
+
+    pub fn address_family(&self) -> SocketAddressFamily {
+        self.family
+    }
+
+    pub fn set_listen_backlog_size(&mut self, value: u32) -> SocketResult<()> {
+        const MIN_BACKLOG: u32 = 1;
+        const MAX_BACKLOG: u32 = i32::MAX as u32; // OS'es will most likely limit it down even further.
+
+        if value == 0 {
+            return Err(ErrorCode::InvalidArgument.into());
+        }
+
+        // Silently clamp backlog size. This is OK for us to do, because operating systems do this too.
+        let value = value.clamp(MIN_BACKLOG, MAX_BACKLOG);
+
+        match &self.tcp_state {
+            TcpState::Default(..) | TcpState::Bound(..) => {
+                // Socket not listening yet. Stash value for first invocation to `listen`.
+            }
+            TcpState::Listening { listener, .. } => {
+                // Try to update the backlog by calling `listen` again.
+                // Not all platforms support this. We'll only update our own value if the OS supports changing the backlog size after the fact.
+
+                rustix::net::listen(&listener, value.try_into().unwrap())
+                    .map_err(|_| ErrorCode::NotSupported)?;
+            }
+            _ => return Err(ErrorCode::InvalidState.into()),
+        }
+        self.listen_backlog_size = value;
+
+        Ok(())
+    }
+
+    pub fn keep_alive_enabled(&self) -> SocketResult<bool> {
+        let view = &*self.as_std_view()?;
+        Ok(sockopt::get_socket_keepalive(view)?)
+    }
+
+    pub fn set_keep_alive_enabled(&self, value: bool) -> SocketResult<()> {
+        let view = &*self.as_std_view()?;
+        Ok(sockopt::set_socket_keepalive(view, value)?)
+    }
+
+    pub fn keep_alive_idle_time(&self) -> SocketResult<std::time::Duration> {
+        let view = &*self.as_std_view()?;
+        Ok(sockopt::get_tcp_keepidle(view)?)
+    }
+
+    pub fn set_keep_alive_idle_time(&mut self, duration: std::time::Duration) -> SocketResult<()> {
+        {
+            let view = &*self.as_std_view()?;
+            network::util::set_tcp_keepidle(view, duration)?;
+        }
+
+        #[cfg(target_os = "macos")]
+        {
+            self.keep_alive_idle_time = Some(duration);
+        }
+
+        Ok(())
+    }
+
+    pub fn keep_alive_interval(&self) -> SocketResult<std::time::Duration> {
+        let view = &*self.as_std_view()?;
+        Ok(sockopt::get_tcp_keepintvl(view)?)
+    }
+
+    pub fn set_keep_alive_interval(&self, duration: std::time::Duration) -> SocketResult<()> {
+        let view = &*self.as_std_view()?;
+        Ok(network::util::set_tcp_keepintvl(view, duration)?)
+    }
+
+    pub fn keep_alive_count(&self) -> SocketResult<u32> {
+        let view = &*self.as_std_view()?;
+        Ok(sockopt::get_tcp_keepcnt(view)?)
+    }
+
+    pub fn set_keep_alive_count(&self, value: u32) -> SocketResult<()> {
+        // TODO(rylev): do we need to check and clamp the value?
+
+        let view = &*self.as_std_view()?;
+        Ok(network::util::set_tcp_keepcnt(view, value)?)
+    }
+
+    pub fn hop_limit(&self) -> SocketResult<u8> {
+        let view = &*self.as_std_view()?;
+
+        let ttl = match self.family {
+            SocketAddressFamily::Ipv4 => network::util::get_ip_ttl(view)?,
+            SocketAddressFamily::Ipv6 => network::util::get_ipv6_unicast_hops(view)?,
+        };
+
+        Ok(ttl)
+    }
+
+    pub fn set_hop_limit(&mut self, value: u8) -> SocketResult<()> {
+        {
+            let view = &*self.as_std_view()?;
+
+            match self.family {
+                SocketAddressFamily::Ipv4 => network::util::set_ip_ttl(view, value)?,
+                SocketAddressFamily::Ipv6 => network::util::set_ipv6_unicast_hops(view, value)?,
+            }
+        }
+
+        #[cfg(target_os = "macos")]
+        {
+            self.hop_limit = Some(value);
+        }
+
+        Ok(())
+    }
+
+    pub fn receive_buffer_size(&self) -> SocketResult<usize> {
+        let view = &*self.as_std_view()?;
+
+        Ok(network::util::get_socket_recv_buffer_size(view)?)
+    }
+
+    pub fn set_receive_buffer_size(&mut self, value: usize) -> SocketResult<()> {
+        {
+            let view = &*self.as_std_view()?;
+
+            network::util::set_socket_recv_buffer_size(view, value)?;
+        }
+
+        #[cfg(target_os = "macos")]
+        {
+            self.receive_buffer_size = Some(value);
+        }
+
+        Ok(())
+    }
+
+    pub fn send_buffer_size(&self) -> SocketResult<usize> {
+        let view = &*self.as_std_view()?;
+
+        Ok(network::util::get_socket_send_buffer_size(view)?)
+    }
+
+    pub fn set_send_buffer_size(&mut self, value: usize) -> SocketResult<()> {
+        {
+            let view = &*self.as_std_view()?;
+
+            network::util::set_socket_send_buffer_size(view, value)?;
+        }
+
+        #[cfg(target_os = "macos")]
+        {
+            self.send_buffer_size = Some(value);
+        }
+
+        Ok(())
+    }
+
+    pub fn shutdown(&self, how: std::net::Shutdown) -> io::Result<()> {
+        let stream = match &self.tcp_state {
+            TcpState::Connected(stream) => stream,
+            _ => {
+                return Err(io::Error::new(
+                    io::ErrorKind::NotConnected,
+                    "socket not connected",
+                ))
+            }
+        };
+
+        stream
+            .as_socketlike_view::<std::net::TcpStream>()
+            .shutdown(how)?;
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl Subscribe for TcpSocket {
+    async fn ready(&mut self) {
+        match &mut self.tcp_state {
+            TcpState::Default(..)
+            | TcpState::BindStarted(..)
+            | TcpState::Bound(..)
+            | TcpState::ListenStarted(..)
+            | TcpState::ConnectReady(..)
+            | TcpState::Closed
+            | TcpState::Connected(..) => {
+                // No async operation in progress.
+            }
+            TcpState::Connecting(future) => {
+                self.tcp_state = TcpState::ConnectReady(future.as_mut().await);
+            }
+            TcpState::Listening {
+                listener,
+                pending_accept,
+            } => match pending_accept {
+                Some(_) => {}
+                None => {
+                    let result = futures::future::poll_fn(|cx| {
+                        listener.poll_accept(cx).map_ok(|(stream, _)| stream)
+                    })
+                    .await;
+                    *pending_accept = Some(result);
+                }
+            },
+        }
+    }
 }
 
 pub(crate) struct TcpReadStream {
@@ -255,94 +845,3 @@ impl Subscribe for TcpWriteStream {
         }
     }
 }
-
-impl TcpSocket {
-    /// Create a new socket in the given family.
-    pub fn new(family: AddressFamily) -> io::Result<Self> {
-        with_ambient_tokio_runtime(|| {
-            let (socket, family) = match family {
-                AddressFamily::Ipv4 => {
-                    let socket = tokio::net::TcpSocket::new_v4()?;
-                    (socket, SocketAddressFamily::Ipv4)
-                }
-                AddressFamily::Ipv6 => {
-                    let socket = tokio::net::TcpSocket::new_v6()?;
-                    sockopt::set_ipv6_v6only(&socket, true)?;
-                    (socket, SocketAddressFamily::Ipv6)
-                }
-            };
-
-            Self::from_state(TcpState::Default(socket), family)
-        })
-    }
-
-    /// Create a `TcpSocket` from an existing socket.
-    pub(crate) fn from_state(state: TcpState, family: SocketAddressFamily) -> io::Result<Self> {
-        Ok(Self {
-            tcp_state: state,
-            listen_backlog_size: DEFAULT_BACKLOG,
-            family,
-            #[cfg(target_os = "macos")]
-            receive_buffer_size: None,
-            #[cfg(target_os = "macos")]
-            send_buffer_size: None,
-            #[cfg(target_os = "macos")]
-            hop_limit: None,
-            #[cfg(target_os = "macos")]
-            keep_alive_idle_time: None,
-        })
-    }
-
-    pub(crate) fn as_std_view(&self) -> SocketResult<SocketlikeView<'_, std::net::TcpStream>> {
-        use crate::bindings::sockets::network::ErrorCode;
-
-        match &self.tcp_state {
-            TcpState::Default(socket) | TcpState::Bound(socket) => {
-                Ok(socket.as_socketlike_view::<std::net::TcpStream>())
-            }
-            TcpState::Connected(stream) => Ok(stream.as_socketlike_view::<std::net::TcpStream>()),
-            TcpState::Listening { listener, .. } => {
-                Ok(listener.as_socketlike_view::<std::net::TcpStream>())
-            }
-
-            TcpState::BindStarted(..)
-            | TcpState::ListenStarted(..)
-            | TcpState::Connecting(..)
-            | TcpState::ConnectReady(..)
-            | TcpState::Closed => Err(ErrorCode::InvalidState.into()),
-        }
-    }
-}
-
-#[async_trait::async_trait]
-impl Subscribe for TcpSocket {
-    async fn ready(&mut self) {
-        match &mut self.tcp_state {
-            TcpState::Default(..)
-            | TcpState::BindStarted(..)
-            | TcpState::Bound(..)
-            | TcpState::ListenStarted(..)
-            | TcpState::ConnectReady(..)
-            | TcpState::Closed
-            | TcpState::Connected(..) => {
-                // No async operation in progress.
-            }
-            TcpState::Connecting(future) => {
-                self.tcp_state = TcpState::ConnectReady(future.as_mut().await);
-            }
-            TcpState::Listening {
-                listener,
-                pending_accept,
-            } => match pending_accept {
-                Some(_) => {}
-                None => {
-                    let result = futures::future::poll_fn(|cx| {
-                        listener.poll_accept(cx).map_ok(|(stream, _)| stream)
-                    })
-                    .await;
-                    *pending_accept = Some(result);
-                }
-            },
-        }
-    }
-}
diff --git a/src/commands/run.rs b/src/commands/run.rs
index 76310b3ad..9040e5a92 100644
--- a/src/commands/run.rs
+++ b/src/commands/run.rs
@@ -26,24 +26,6 @@ use wasmtime_wasi_threads::WasiThreadsCtx;
 #[cfg(feature = "wasi-http")]
 use wasmtime_wasi_http::WasiHttpCtx;
 
-fn parse_env_var(s: &str) -> Result<(String, Option<String>)> {
-    let mut parts = s.splitn(2, '=');
-    Ok((
-        parts.next().unwrap().to_string(),
-        parts.next().map(|s| s.to_string()),
-    ))
-}
-
-fn parse_dirs(s: &str) -> Result<(String, String)> {
-    let mut parts = s.split("::");
-    let host = parts.next().unwrap();
-    let guest = match parts.next() {
-        Some(guest) => guest,
-        None => host,
-    };
-    Ok((host.into(), guest.into()))
-}
-
 fn parse_preloads(s: &str) -> Result<(String, PathBuf)> {
     let parts: Vec<&str> = s.splitn(2, '=').collect();
     if parts.len() != 2 {
@@ -59,25 +41,6 @@ pub struct RunCommand {
     #[allow(missing_docs)]
     pub run: RunCommon,
 
-    /// Grant access of a host directory to a guest.
-    ///
-    /// If specified as just `HOST_DIR` then the same directory name on the
-    /// host is made available within the guest. If specified as `HOST::GUEST`
-    /// then the `HOST` directory is opened and made available as the name
-    /// `GUEST` in the guest.
-    #[arg(long = "dir", value_name = "HOST_DIR[::GUEST_DIR]", value_parser = parse_dirs)]
-    pub dirs: Vec<(String, String)>,
-
-    /// Pass an environment variable to the program.
-    ///
-    /// The `--env FOO=BAR` form will set the environment variable named `FOO`
-    /// to the value `BAR` for the guest program using WASI. The `--env FOO`
-    /// form will set the environment variable named `FOO` to the same value it
-    /// has in the calling process for the guest, or in other words it will
-    /// cause the environment variable `FOO` to be inherited.
-    #[arg(long = "env", number_of_values = 1, value_name = "NAME[=VAL]", value_parser = parse_env_var)]
-    pub vars: Vec<(String, Option<String>)>,
-
     /// The name of the function to run
     #[arg(long, value_name = "FUNCTION")]
     pub invoke: Option<String>,
@@ -263,20 +226,6 @@ impl RunCommand {
         Ok(())
     }
 
-    fn compute_preopen_sockets(&self) -> Result<Vec<TcpListener>> {
-        let mut listeners = vec![];
-
-        for address in &self.run.common.wasi.tcplisten {
-            let stdlistener = std::net::TcpListener::bind(address)
-                .with_context(|| format!("failed to bind to address '{}'", address))?;
-
-            let _ = stdlistener.set_nonblocking(true)?;
-
-            listeners.push(TcpListener::from_std(stdlistener))
-        }
-        Ok(listeners)
-    }
-
     fn compute_argv(&self) -> Result<Vec<String>> {
         let mut result = Vec::new();
 
@@ -762,7 +711,7 @@ impl RunCommand {
                 builder.env(&k, &v)?;
             }
         }
-        for (key, value) in self.vars.iter() {
+        for (key, value) in self.run.vars.iter() {
             let value = match value {
                 Some(value) => value.clone(),
                 None => match std::env::var_os(key) {
@@ -784,12 +733,13 @@ impl RunCommand {
             num_fd = ctx_set_listenfd(num_fd, &mut builder)?;
         }
 
-        for listener in self.compute_preopen_sockets()? {
+        for listener in self.run.compute_preopen_sockets()? {
+            let listener = TcpListener::from_std(listener);
             builder.preopened_socket(num_fd as _, listener)?;
             num_fd += 1;
         }
 
-        for (host, guest) in self.dirs.iter() {
+        for (host, guest) in self.run.dirs.iter() {
             let dir = Dir::open_ambient_dir(host, ambient_authority())
                 .with_context(|| format!("failed to open directory '{}'", host))?;
             builder.preopened_dir(dir, guest)?;
@@ -802,63 +752,7 @@ impl RunCommand {
     fn set_preview2_ctx(&self, store: &mut Store<Host>) -> Result<()> {
         let mut builder = wasmtime_wasi::WasiCtxBuilder::new();
         builder.inherit_stdio().args(&self.compute_argv()?);
-
-        // It's ok to block the current thread since we're the only thread in
-        // the program as the CLI. This helps improve the performance of some
-        // blocking operations in WASI, for example, by skipping the
-        // back-and-forth between sync and async.
-        builder.allow_blocking_current_thread(true);
-
-        if self.run.common.wasi.inherit_env == Some(true) {
-            for (k, v) in std::env::vars() {
-                builder.env(&k, &v);
-            }
-        }
-        for (key, value) in self.vars.iter() {
-            let value = match value {
-                Some(value) => value.clone(),
-                None => match std::env::var_os(key) {
-                    Some(val) => val
-                        .into_string()
-                        .map_err(|_| anyhow!("environment variable `{key}` not valid utf-8"))?,
-                    None => {
-                        // leave the env var un-set in the guest
-                        continue;
-                    }
-                },
-            };
-            builder.env(key, &value);
-        }
-
-        if self.run.common.wasi.listenfd == Some(true) {
-            bail!("components do not support --listenfd");
-        }
-        for _ in self.compute_preopen_sockets()? {
-            bail!("components do not support --tcplisten");
-        }
-
-        for (host, guest) in self.dirs.iter() {
-            builder.preopened_dir(
-                host,
-                guest,
-                wasmtime_wasi::DirPerms::all(),
-                wasmtime_wasi::FilePerms::all(),
-            )?;
-        }
-
-        if self.run.common.wasi.inherit_network == Some(true) {
-            builder.inherit_network();
-        }
-        if let Some(enable) = self.run.common.wasi.allow_ip_name_lookup {
-            builder.allow_ip_name_lookup(enable);
-        }
-        if let Some(enable) = self.run.common.wasi.tcp {
-            builder.allow_tcp(enable);
-        }
-        if let Some(enable) = self.run.common.wasi.udp {
-            builder.allow_udp(enable);
-        }
-
+        self.run.configure_wasip2(&mut builder)?;
         let ctx = builder.build_p1();
         store.data_mut().preview2_ctx = Some(Arc::new(Mutex::new(ctx)));
         Ok(())
diff --git a/src/commands/serve.rs b/src/commands/serve.rs
index 22520e28b..5e7e7f0bb 100644
--- a/src/commands/serve.rs
+++ b/src/commands/serve.rs
@@ -132,8 +132,9 @@ impl ServeCommand {
 
     fn new_store(&self, engine: &Engine, req_id: u64) -> Result<Store<Host>> {
         let mut builder = WasiCtxBuilder::new();
+        self.run.configure_wasip2(&mut builder)?;
 
-        builder.envs(&[("REQUEST_ID", req_id.to_string())]);
+        builder.env("REQUEST_ID", req_id.to_string());
 
         builder.stdout(LogStream {
             prefix: format!("stdout [{req_id}] :: "),
diff --git a/src/common.rs b/src/common.rs
index 3257fca92..4cbd70f32 100644
--- a/src/common.rs
+++ b/src/common.rs
@@ -1,10 +1,12 @@
 //! Common functionality shared between command implementations.
 
-use anyhow::{bail, Context, Result};
+use anyhow::{anyhow, bail, Context, Result};
 use clap::Parser;
+use std::net::TcpListener;
 use std::{path::Path, time::Duration};
 use wasmtime::{Engine, Module, Precompiled, StoreLimits, StoreLimitsBuilder};
 use wasmtime_cli_flags::{opt::WasmtimeOptionValue, CommonOptions};
+use wasmtime_wasi::WasiCtxBuilder;
 
 #[cfg(feature = "component-model")]
 use wasmtime::component::Component;
@@ -70,6 +72,43 @@ pub struct RunCommon {
         value_parser = Profile::parse,
     )]
     pub profile: Option<Profile>,
+
+    /// Grant access of a host directory to a guest.
+    ///
+    /// If specified as just `HOST_DIR` then the same directory name on the
+    /// host is made available within the guest. If specified as `HOST::GUEST`
+    /// then the `HOST` directory is opened and made available as the name
+    /// `GUEST` in the guest.
+    #[arg(long = "dir", value_name = "HOST_DIR[::GUEST_DIR]", value_parser = parse_dirs)]
+    pub dirs: Vec<(String, String)>,
+
+    /// Pass an environment variable to the program.
+    ///
+    /// The `--env FOO=BAR` form will set the environment variable named `FOO`
+    /// to the value `BAR` for the guest program using WASI. The `--env FOO`
+    /// form will set the environment variable named `FOO` to the same value it
+    /// has in the calling process for the guest, or in other words it will
+    /// cause the environment variable `FOO` to be inherited.
+    #[arg(long = "env", number_of_values = 1, value_name = "NAME[=VAL]", value_parser = parse_env_var)]
+    pub vars: Vec<(String, Option<String>)>,
+}
+
+fn parse_env_var(s: &str) -> Result<(String, Option<String>)> {
+    let mut parts = s.splitn(2, '=');
+    Ok((
+        parts.next().unwrap().to_string(),
+        parts.next().map(|s| s.to_string()),
+    ))
+}
+
+fn parse_dirs(s: &str) -> Result<(String, String)> {
+    let mut parts = s.split("::");
+    let host = parts.next().unwrap();
+    let guest = match parts.next() {
+        Some(guest) => guest,
+        None => host,
+    };
+    Ok((host.into(), guest.into()))
 }
 
 impl RunCommon {
@@ -222,6 +261,80 @@ impl RunCommon {
             }
         })
     }
+
+    pub fn configure_wasip2(&self, builder: &mut WasiCtxBuilder) -> Result<()> {
+        // It's ok to block the current thread since we're the only thread in
+        // the program as the CLI. This helps improve the performance of some
+        // blocking operations in WASI, for example, by skipping the
+        // back-and-forth between sync and async.
+        builder.allow_blocking_current_thread(true);
+
+        if self.common.wasi.inherit_env == Some(true) {
+            for (k, v) in std::env::vars() {
+                builder.env(&k, &v);
+            }
+        }
+        for (key, value) in self.vars.iter() {
+            let value = match value {
+                Some(value) => value.clone(),
+                None => match std::env::var_os(key) {
+                    Some(val) => val
+                        .into_string()
+                        .map_err(|_| anyhow!("environment variable `{key}` not valid utf-8"))?,
+                    None => {
+                        // leave the env var un-set in the guest
+                        continue;
+                    }
+                },
+            };
+            builder.env(key, &value);
+        }
+
+        for (host, guest) in self.dirs.iter() {
+            builder.preopened_dir(
+                host,
+                guest,
+                wasmtime_wasi::DirPerms::all(),
+                wasmtime_wasi::FilePerms::all(),
+            )?;
+        }
+
+        if self.common.wasi.listenfd == Some(true) {
+            bail!("components do not support --listenfd");
+        }
+        for _ in self.compute_preopen_sockets()? {
+            bail!("components do not support --tcplisten");
+        }
+
+        if self.common.wasi.inherit_network == Some(true) {
+            builder.inherit_network();
+        }
+        if let Some(enable) = self.common.wasi.allow_ip_name_lookup {
+            builder.allow_ip_name_lookup(enable);
+        }
+        if let Some(enable) = self.common.wasi.tcp {
+            builder.allow_tcp(enable);
+        }
+        if let Some(enable) = self.common.wasi.udp {
+            builder.allow_udp(enable);
+        }
+
+        Ok(())
+    }
+
+    pub fn compute_preopen_sockets(&self) -> Result<Vec<TcpListener>> {
+        let mut listeners = vec![];
+
+        for address in &self.common.wasi.tcplisten {
+            let stdlistener = std::net::TcpListener::bind(address)
+                .with_context(|| format!("failed to bind to address '{}'", address))?;
+
+            let _ = stdlistener.set_nonblocking(true)?;
+
+            listeners.push(stdlistener)
+        }
+        Ok(listeners)
+    }
 }
 
 #[derive(Clone, PartialEq)]
diff --git a/src/old_cli.rs b/src/old_cli.rs
index c863fc1fd..7886d5ffc 100644
--- a/src/old_cli.rs
+++ b/src/old_cli.rs
@@ -816,14 +816,14 @@ impl RunCommand {
             common,
             allow_precompiled,
             profile: profile.map(|p| p.convert()),
+            dirs,
+            vars,
         };
 
         let mut module_and_args = vec![module.into()];
         module_and_args.extend(module_args.into_iter().map(|s| s.into()));
         crate::commands::RunCommand {
             run,
-            dirs,
-            vars,
             invoke,
             preloads,
             module_and_args,
diff --git a/supply-chain/audits.toml b/supply-chain/audits.toml
index 3773ed23d..0c5195ca7 100644
--- a/supply-chain/audits.toml
+++ b/supply-chain/audits.toml
@@ -3727,6 +3727,12 @@ user-id = 73222 # wasmtime-publish
 start = "2024-02-15"
 end = "2025-02-16"
 
+[[trusted.wasmtime-slab]]
+criteria = "safe-to-deploy"
+user-id = 73222 # wasmtime-publish
+start = "2024-03-20"
+end = "2025-04-02"
+
 [[trusted.wasmtime-wmemcheck]]
 criteria = "safe-to-deploy"
 user-id = 73222 # wasmtime-publish
diff --git a/supply-chain/config.toml b/supply-chain/config.toml
index 369826159..66021af73 100644
--- a/supply-chain/config.toml
+++ b/supply-chain/config.toml
@@ -136,6 +136,9 @@ audit-as-crates-io = true
 [policy.wasmtime-runtime]
 audit-as-crates-io = true
 
+[policy.wasmtime-slab]
+audit-as-crates-io = true
+
 [policy.wasmtime-types]
 audit-as-crates-io = true
 
diff --git a/supply-chain/imports.lock b/supply-chain/imports.lock
index a4490981e..5d7c3555e 100644
--- a/supply-chain/imports.lock
+++ b/supply-chain/imports.lock
@@ -133,6 +133,10 @@ audited_as = "18.0.2"
 version = "20.0.0"
 audited_as = "18.0.2"
 
+[[unpublished.wasmtime-slab]]
+version = "20.0.0"
+audited_as = "19.0.0"
+
 [[unpublished.wasmtime-types]]
 version = "20.0.0"
 audited_as = "18.0.2"
@@ -1026,6 +1030,12 @@ when = "2024-02-28"
 user-id = 73222
 user-login = "wasmtime-publish"
 
+[[publisher.wasmtime-slab]]
+version = "19.0.0"
+when = "2024-03-20"
+user-id = 73222
+user-login = "wasmtime-publish"
+
 [[publisher.wasmtime-types]]
 version = "18.0.2"
 when = "2024-02-28"
diff --git a/winch/codegen/src/isa/mod.rs b/winch/codegen/src/isa/mod.rs
index e34dd5e2b..b2800c769 100644
--- a/winch/codegen/src/isa/mod.rs
+++ b/winch/codegen/src/isa/mod.rs
@@ -181,7 +181,7 @@ pub trait TargetIsa: Send + Sync {
         }
     }
 
-    /// Get the endianess of the underlying target triple.
+    /// Get the endianness of the underlying target triple.
     fn endianness(&self) -> target_lexicon::Endianness {
         self.triple().endianness().unwrap()
     }
diff --git a/winch/codegen/src/masm.rs b/winch/codegen/src/masm.rs
index f2ec4cadc..e4a69ab8d 100644
--- a/winch/codegen/src/masm.rs
+++ b/winch/codegen/src/masm.rs
@@ -436,7 +436,7 @@ pub const TRUSTED_FLAGS: MemFlags = MemFlags::trusted();
 
 /// Flags used for WebAssembly loads / stores.
 /// Untrusted by default so we don't set `no_trap`.
-/// We also ensure that the endianess is the right one for WebAssembly.
+/// We also ensure that the endianness is the right one for WebAssembly.
 pub const UNTRUSTED_FLAGS: MemFlags = MemFlags::new().with_endianness(Endianness::Little);
 
 /// Generic MacroAssembler interface used by the code generation.
@@ -554,7 +554,7 @@ pub(crate) trait MacroAssembler {
     /// [Self::store], more precisely, it can implicitly trap, in certain
     /// circumstances, even if explicit bounds checks are elided, in that sense,
     /// we consider this type of load as untrusted. It can also differ with
-    /// regards to the endianess depending on the target ISA. For this reason,
+    /// regards to the endianness depending on the target ISA. For this reason,
     /// [Self::wasm_store], should be explicitly used when emitting WebAssembly
     /// stores.
     fn wasm_store(&mut self, src: Reg, dst: Self::Address, size: OperandSize);
@@ -567,7 +567,7 @@ pub(crate) trait MacroAssembler {
     /// [Self::load], more precisely, it can implicitly trap, in certain
     /// circumstances, even if explicit bounds checks are elided, in that sense,
     /// we consider this type of load as untrusted. It can also differ with
-    /// regards to the endianess depending on the target ISA. For this reason,
+    /// regards to the endianness depending on the target ISA. For this reason,
     /// [Self::wasm_load], should be explicitly used when emitting WebAssembly
     /// loads.
     fn wasm_load(


Test patch:
diff --git a/crates/test-programs/src/bin/api_proxy.rs b/crates/test-programs/src/bin/api_proxy.rs
index 28166a2af..bbbbecd1e 100644
--- a/crates/test-programs/src/bin/api_proxy.rs
+++ b/crates/test-programs/src/bin/api_proxy.rs
@@ -1,18 +1,12 @@
-pub mod bindings {
-    wit_bindgen::generate!({
-        path: "../wasi-http/wit",
-        world: "wasi:http/proxy",
-        default_bindings_module: "bindings",
-    });
-}
-
-use bindings::wasi::http::types::{IncomingRequest, ResponseOutparam};
+use test_programs::wasi::http::types::{
+    Headers, IncomingRequest, OutgoingBody, OutgoingResponse, ResponseOutparam,
+};
 
 struct T;
 
-bindings::export!(T);
+test_programs::proxy::export!(T);
 
-impl bindings::exports::wasi::http::incoming_handler::Guest for T {
+impl test_programs::proxy::exports::wasi::http::incoming_handler::Guest for T {
     fn handle(request: IncomingRequest, outparam: ResponseOutparam) {
         assert!(request.scheme().is_some());
         assert!(request.authority().is_some());
@@ -41,19 +35,18 @@ impl bindings::exports::wasi::http::incoming_handler::Guest for T {
             "forbidden host header present in incoming request"
         );
 
-        let hdrs = bindings::wasi::http::types::Headers::new();
-        let resp = bindings::wasi::http::types::OutgoingResponse::new(hdrs);
+        let hdrs = Headers::new();
+        let resp = OutgoingResponse::new(hdrs);
         let body = resp.body().expect("outgoing response");
 
-        bindings::wasi::http::types::ResponseOutparam::set(outparam, Ok(resp));
+        ResponseOutparam::set(outparam, Ok(resp));
 
         let out = body.write().expect("outgoing stream");
         out.blocking_write_and_flush(b"hello, world!")
             .expect("writing response");
 
         drop(out);
-        bindings::wasi::http::types::OutgoingBody::finish(body, None)
-            .expect("outgoing-body.finish");
+        OutgoingBody::finish(body, None).expect("outgoing-body.finish");
     }
 }
 
diff --git a/crates/test-programs/src/bin/api_proxy_streaming.rs b/crates/test-programs/src/bin/api_proxy_streaming.rs
index 689247908..328e8a12d 100644
--- a/crates/test-programs/src/bin/api_proxy_streaming.rs
+++ b/crates/test-programs/src/bin/api_proxy_streaming.rs
@@ -1,26 +1,18 @@
 use anyhow::{anyhow, bail, Result};
-use bindings::wasi::http::types::{
+use futures::{future, stream, Future, SinkExt, StreamExt, TryStreamExt};
+use test_programs::wasi::http::types::{
     Fields, IncomingRequest, IncomingResponse, Method, OutgoingBody, OutgoingRequest,
     OutgoingResponse, ResponseOutparam, Scheme,
 };
-use futures::{future, stream, Future, SinkExt, StreamExt, TryStreamExt};
 use url::Url;
 
-mod bindings {
-    wit_bindgen::generate!({
-        path: "../wasi-http/wit",
-        world: "wasi:http/proxy",
-        default_bindings_module: "bindings",
-    });
-}
-
 const MAX_CONCURRENCY: usize = 16;
 
 struct Handler;
 
-bindings::export!(Handler);
+test_programs::proxy::export!(Handler);
 
-impl bindings::exports::wasi::http::incoming_handler::Guest for Handler {
+impl test_programs::proxy::exports::wasi::http::incoming_handler::Guest for Handler {
     fn handle(request: IncomingRequest, response_out: ResponseOutparam) {
         executor::run(async move {
             handle_request(request, response_out).await;
@@ -312,16 +304,6 @@ async fn hash(url: &Url) -> Result<String> {
 fn main() {}
 
 mod executor {
-    use super::bindings::wasi::{
-        http::{
-            outgoing_handler,
-            types::{
-                self, IncomingBody, IncomingResponse, InputStream, OutgoingBody, OutgoingRequest,
-                OutputStream,
-            },
-        },
-        io::{self, streams::StreamError},
-    };
     use anyhow::{anyhow, Error, Result};
     use futures::{future, sink, stream, Sink, Stream};
     use std::{
@@ -332,6 +314,16 @@ mod executor {
         sync::{Arc, Mutex},
         task::{Context, Poll, Wake, Waker},
     };
+    use test_programs::wasi::{
+        http::{
+            outgoing_handler,
+            types::{
+                self, IncomingBody, IncomingResponse, InputStream, OutgoingBody, OutgoingRequest,
+                OutputStream,
+            },
+        },
+        io::{self, streams::StreamError},
+    };
 
     const READ_SIZE: u64 = 16 * 1024;
 
diff --git a/crates/test-programs/src/bin/cli_serve_echo_env.rs b/crates/test-programs/src/bin/cli_serve_echo_env.rs
new file mode 100644
index 000000000..94294cc0a
--- /dev/null
+++ b/crates/test-programs/src/bin/cli_serve_echo_env.rs
@@ -0,0 +1,26 @@
+use test_programs::proxy;
+use test_programs::wasi::http::types::{
+    Fields, IncomingRequest, OutgoingResponse, ResponseOutparam,
+};
+
+struct T;
+
+proxy::export!(T);
+
+impl proxy::exports::wasi::http::incoming_handler::Guest for T {
+    fn handle(request: IncomingRequest, outparam: ResponseOutparam) {
+        let headers = request.headers();
+        let header_key = "env".to_string();
+        let env_var = headers.get(&header_key);
+        assert!(env_var.len() == 1, "should have exactly one `env` header");
+        let key = std::str::from_utf8(&env_var[0]).unwrap();
+        let fields = Fields::new();
+        if let Ok(val) = std::env::var(key) {
+            fields.set(&header_key, &[val.into_bytes()]).unwrap();
+        }
+        let resp = OutgoingResponse::new(fields);
+        ResponseOutparam::set(outparam, Ok(resp));
+    }
+}
+
+fn main() {}
diff --git a/crates/test-programs/src/lib.rs b/crates/test-programs/src/lib.rs
index 25ed50efb..9697cb631 100644
--- a/crates/test-programs/src/lib.rs
+++ b/crates/test-programs/src/lib.rs
@@ -3,3 +3,16 @@ pub mod preview1;
 pub mod sockets;
 
 wit_bindgen::generate!("test-command" in "../wasi/wit");
+
+pub mod proxy {
+    wit_bindgen::generate!({
+        path: "../wasi-http/wit",
+        world: "wasi:http/proxy",
+        default_bindings_module: "test_programs::proxy",
+        pub_export_macro: true,
+        with: {
+            "wasi:http/types@0.2.0": crate::wasi::http::types,
+            "wasi:http/outgoing-handler@0.2.0": crate::wasi::http::outgoing_handler,
+        },
+    });
+}
diff --git a/tests/all/cli_tests.rs b/tests/all/cli_tests.rs
index 3a604b4f0..930e709b4 100644
--- a/tests/all/cli_tests.rs
+++ b/tests/all/cli_tests.rs
@@ -1260,10 +1260,14 @@ fn mpk_without_pooling() -> Result<()> {
 
 mod test_programs {
     use super::{get_wasmtime_command, run_wasmtime};
-    use anyhow::Result;
-    use std::io::{Read, Write};
-    use std::process::Stdio;
+    use anyhow::{bail, Context, Result};
+    use http_body_util::BodyExt;
+    use hyper::header::HeaderValue;
+    use std::io::{BufRead, BufReader, Read, Write};
+    use std::net::SocketAddr;
+    use std::process::{Child, Command, Stdio};
     use test_programs_artifacts::*;
+    use tokio::net::TcpStream;
 
     macro_rules! assert_test_exists {
         ($name:ident) => {
@@ -1603,6 +1607,174 @@ mod test_programs {
         run_wasmtime(&["run", CLI_SLEEP_COMPONENT])?;
         Ok(())
     }
+
+    /// Helper structure to manage an invocation of `wasmtime serve`
+    struct WasmtimeServe {
+        child: Option<Child>,
+        addr: SocketAddr,
+    }
+
+    impl WasmtimeServe {
+        /// Creates a new server which will serve the wasm component pointed to
+        /// by `wasm`.
+        ///
+        /// A `configure` callback is provided to specify how `wasmtime serve`
+        /// will be invoked and configure arguments such as headers.
+        fn new(wasm: &str, configure: impl FnOnce(&mut Command)) -> Result<WasmtimeServe> {
+            // Spawn `wasmtime serve` on port 0 which will randomly assign it a
+            // port.
+            let mut cmd = super::get_wasmtime_command()?;
+            cmd.stdin(Stdio::null());
+            cmd.stdout(Stdio::piped());
+            cmd.stderr(Stdio::piped());
+            cmd.arg("serve").arg("--addr=127.0.0.1:0").arg(wasm);
+            configure(&mut cmd);
+            let mut child = cmd.spawn()?;
+
+            // Read the first line of stderr which will say which address it's
+            // listening on.
+            //
+            // NB: this intentionally discards any extra buffered data in the
+            // `BufReader` once the newline is found. The server shouldn't print
+            // anything interesting other than the address so once we get a line
+            // all remaining output is left to be captured by future requests
+            // send to the server.
+            let mut line = String::new();
+            BufReader::new(child.stderr.as_mut().unwrap()).read_line(&mut line)?;
+            let addr_start = line.find("127.0.0.1").unwrap();
+            let addr = &line[addr_start..];
+            let addr_end = addr.find("/").unwrap();
+            let addr = &addr[..addr_end];
+            Ok(WasmtimeServe {
+                child: Some(child),
+                addr: addr.parse().unwrap(),
+            })
+        }
+
+        /// Completes this server gracefully by printing the output on failure.
+        fn finish(mut self) -> Result<()> {
+            let mut child = self.child.take().unwrap();
+
+            // If the child process has already exited then collect the output
+            // and test if it succeeded. Otherwise it's still running so kill it
+            // and then reap it. Assume that if it's still running then the test
+            // has otherwise passed so no need to print the output.
+            if child.try_wait()?.is_some() {
+                let output = child.wait_with_output()?;
+                if output.status.success() {
+                    return Ok(());
+                }
+                bail!("child failed {output:?}");
+            } else {
+                child.kill()?;
+                child.wait_with_output()?;
+            }
+            Ok(())
+        }
+
+        /// Send a request to this server and wait for the response.
+        async fn send_request(&self, req: http::Request<String>) -> Result<http::Response<String>> {
+            let tcp = TcpStream::connect(&self.addr)
+                .await
+                .context("failed to connect")?;
+            let tcp = wasmtime_wasi_http::io::TokioIo::new(tcp);
+            let (mut send, conn) = hyper::client::conn::http1::handshake(tcp)
+                .await
+                .context("failed http handshake")?;
+
+            let conn_task = tokio::task::spawn(conn);
+
+            let response = send
+                .send_request(req)
+                .await
+                .context("error sending request")?;
+            drop(send);
+            let (parts, body) = response.into_parts();
+
+            let body = body.collect().await.context("failed to read body")?;
+            assert!(body.trailers().is_none());
+            let body = std::str::from_utf8(&body.to_bytes())?.to_string();
+
+            conn_task.await??;
+
+            Ok(http::Response::from_parts(parts, body))
+        }
+    }
+
+    // Don't leave child processes running by accident so kill the child process
+    // if our server goes away.
+    impl Drop for WasmtimeServe {
+        fn drop(&mut self) {
+            let mut child = match self.child.take() {
+                Some(child) => child,
+                None => return,
+            };
+            if child.kill().is_err() {
+                return;
+            }
+            let output = match child.wait_with_output() {
+                Ok(output) => output,
+                Err(_) => return,
+            };
+
+            println!("server status: {}", output.status);
+            if !output.stdout.is_empty() {
+                println!(
+                    "server stdout:\n{}",
+                    String::from_utf8_lossy(&output.stdout)
+                );
+            }
+            if !output.stderr.is_empty() {
+                println!(
+                    "server stderr:\n{}",
+                    String::from_utf8_lossy(&output.stderr)
+                );
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn cli_serve_echo_env() -> Result<()> {
+        let server = WasmtimeServe::new(CLI_SERVE_ECHO_ENV_COMPONENT, |cmd| {
+            cmd.arg("--env=FOO=bar");
+            cmd.arg("--env=BAR");
+            cmd.arg("-Scli");
+            cmd.env_remove("BAR");
+        })?;
+
+        let foo_env = server
+            .send_request(
+                hyper::Request::builder()
+                    .uri("http://localhost/")
+                    .header("env", "FOO")
+                    .body(String::new())
+                    .context("failed to make request")?,
+            )
+            .await?;
+
+        assert!(foo_env.status().is_success());
+        assert!(foo_env.body().is_empty());
+        let headers = foo_env.headers();
+        assert_eq!(headers.get("env"), Some(&HeaderValue::from_static("bar")));
+
+        let bar_env = server
+            .send_request(
+                hyper::Request::builder()
+                    .uri("http://localhost/")
+                    .header("env", "BAR")
+                    .body(String::new())
+                    .context("failed to make request")?,
+            )
+            .await?;
+
+        assert!(bar_env.status().is_success());
+        assert!(bar_env.body().is_empty());
+        let headers = bar_env.headers();
+        assert_eq!(headers.get("env"), None);
+
+        server.finish()?;
+        Ok(())
+    }
 }
 
 #[test]
diff --git a/tests/misc_testsuite/externref-table-dropped-segment-issue-8281.wast b/tests/misc_testsuite/externref-table-dropped-segment-issue-8281.wast
new file mode 100644
index 000000000..723c03c80
--- /dev/null
+++ b/tests/misc_testsuite/externref-table-dropped-segment-issue-8281.wast
@@ -0,0 +1,39 @@
+(module
+  (table $t 0 0 externref)
+
+  (func (export "f1")
+    (i32.const 0)
+    (i32.const 0)
+    (i32.const 0)
+    (table.init $t $declared)
+  )
+
+  (func (export "f2")
+    (i32.const 0)
+    (i32.const 0)
+    (i32.const 0)
+    (table.init $t $passive)
+
+    (elem.drop $passive)
+
+    (i32.const 0)
+    (i32.const 0)
+    (i32.const 0)
+    (table.init $t $passive)
+  )
+
+  (func (export "f3")
+    (i32.const 0)
+    (i32.const 0)
+    (i32.const 0)
+    (table.init $t $active)
+  )
+
+  (elem $declared declare externref)
+  (elem $passive externref)
+  (elem $active (i32.const 0) externref)
+)
+
+(assert_return (invoke "f1"))
+(assert_return (invoke "f2"))
+(assert_return (invoke "f3"))