Repository: bytecodealliance/wasmtime
Pull request #9231 (2024-09-12 15:00:39)

Revert back to 1.78 as workspace MSRV
Last night's OSS-Fuzz build failed because the compiler used there hasn't been updated in quite some time. It's a nightly of 1.78 from the beginning of this year which means the bump to 1.79 recently for this workspace meant that Wasmtime stopped building there.

This commit reverts ac607ed2abdf0d51aedca302a78aa27d03451d2c and 1ad3da85ef8d161071b57aeed86effdd70cbc1ce and then updates our github action to say that we're 3 versions behind the latest Rust right now with some commentary as to why.

<!--
Please make sure you include the following information:

- If this work has been discussed elsewhere, please include a link to that
  conversation. If it was discussed in an issue, just mention "issue #...".

- Explain why this change is needed. If the details are in an issue already,
  this can be brief.

Our development process is documented in the Wasmtime book:
https://docs.wasmtime.dev/contributing-development-process.html

Please ensure all communication follows the code of conduct:
https://github.com/bytecodealliance/wasmtime/blob/main/CODE_OF_CONDUCT.md
-->



Patch:
diff --git a/.github/actions/install-rust/action.yml b/.github/actions/install-rust/action.yml
index 55ba39879..e0ebe61c4 100644
--- a/.github/actions/install-rust/action.yml
+++ b/.github/actions/install-rust/action.yml
@@ -6,6 +6,18 @@ inputs:
     description: 'Default toolchan to install'
     required: false
     default: 'default'
+  msrv_range:
+    description: 'Versions later-than-latest-Rust the MSRV supports'
+    required: false
+    # Note that this is currently set to 3 as the MSRV is 1.78 and current
+    # stable is 1.81. Currently MSRV cannot be updated as it would break the
+    # build on OSS-Fuzz which hasn't updated its Rust compiler in quite some time.
+    #
+    # Updating OSS-Fuzz is being done in
+    # https://github.com/google/oss-fuzz/pull/12365 but it's taking some time,
+    # so for now this'll get bumped instead of MSRV while we wait for the update
+    # to happen.
+    default: '3'
 
 runs:
   using: composite
@@ -17,9 +29,10 @@ runs:
         # Determine MSRV as N in `1.N.0` by looking at the `rust-version`
         # located in the root `Cargo.toml`.
         msrv=$(grep 'rust-version.*1' Cargo.toml | sed 's/.*\.\([0-9]*\)\..*/\1/')
+        range=${{ inputs.msrv_range }}
 
         if [ "${{ inputs.toolchain }}" = "default" ]; then
-          echo "version=1.$((msrv+2)).0" >> "$GITHUB_OUTPUT"
+          echo "version=1.$((msrv+range)).0" >> "$GITHUB_OUTPUT"
         elif [ "${{ inputs.toolchain }}" = "msrv" ]; then
           echo "version=1.$msrv.0" >> "$GITHUB_OUTPUT"
         elif [ "${{ inputs.toolchain }}" = "wasmtime-ci-pinned-nightly" ]; then
diff --git a/Cargo.toml b/Cargo.toml
index abe433b4e..32eca81f4 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -161,7 +161,7 @@ authors = ["The Wasmtime Project Developers"]
 edition = "2021"
 # Wasmtime's current policy is that this number can be no larger than the
 # current stable release of Rust minus 2.
-rust-version = "1.79.0"
+rust-version = "1.78.0"
 
 [workspace.lints.rust]
 # Turn on some lints which are otherwise allow-by-default in rustc.
diff --git a/crates/wasi/src/filesystem.rs b/crates/wasi/src/filesystem.rs
index a924fa96a..2c656b699 100644
--- a/crates/wasi/src/filesystem.rs
+++ b/crates/wasi/src/filesystem.rs
@@ -361,7 +361,7 @@ impl HostInputStream for FileInputStream {
                 // background tasks. Also, the guest is only blocked if
                 // the stream was dropped mid-read, which we don't expect to
                 // occur frequently.
-                task.abort_wait().await;
+                task.cancel().await;
             }
             _ => {}
         }
@@ -558,7 +558,7 @@ impl HostOutputStream for FileOutputStream {
                 // background tasks. Also, the guest is only blocked if
                 // the stream was dropped mid-write, which we don't expect to
                 // occur frequently.
-                task.abort_wait().await;
+                task.cancel().await;
             }
             _ => {}
         }
diff --git a/crates/wasi/src/host/tcp.rs b/crates/wasi/src/host/tcp.rs
index 2f20cdc25..143ea607e 100644
--- a/crates/wasi/src/host/tcp.rs
+++ b/crates/wasi/src/host/tcp.rs
@@ -298,7 +298,7 @@ where
             ShutdownType::Send => std::net::Shutdown::Write,
             ShutdownType::Both => std::net::Shutdown::Both,
         };
-        Ok(socket.shutdown(how)?)
+        socket.shutdown(how)
     }
 
     fn drop(&mut self, this: Resource<tcp::TcpSocket>) -> Result<(), anyhow::Error> {
diff --git a/crates/wasi/src/pipe.rs b/crates/wasi/src/pipe.rs
index 4b723cc0c..83a389d90 100644
--- a/crates/wasi/src/pipe.rs
+++ b/crates/wasi/src/pipe.rs
@@ -193,7 +193,7 @@ impl HostInputStream for AsyncReadStream {
 
     async fn cancel(&mut self) {
         match self.join_handle.take() {
-            Some(task) => _ = task.abort_wait().await,
+            Some(task) => _ = task.cancel().await,
             None => {}
         }
     }
diff --git a/crates/wasi/src/runtime.rs b/crates/wasi/src/runtime.rs
index dc29fef97..b1fb78396 100644
--- a/crates/wasi/src/runtime.rs
+++ b/crates/wasi/src/runtime.rs
@@ -42,8 +42,9 @@ pub struct AbortOnDropJoinHandle<T>(tokio::task::JoinHandle<T>);
 impl<T> AbortOnDropJoinHandle<T> {
     /// Abort the task and wait for it to finish. Optionally returns the result
     /// of the task if it ran to completion prior to being aborted.
-    pub(crate) async fn abort_wait(mut self) -> Option<T> {
+    pub(crate) async fn cancel(mut self) -> Option<T> {
         self.0.abort();
+
         match (&mut self.0).await {
             Ok(value) => Some(value),
             Err(err) if err.is_cancelled() => None,
diff --git a/crates/wasi/src/tcp.rs b/crates/wasi/src/tcp.rs
index 32feef95a..44ba68416 100644
--- a/crates/wasi/src/tcp.rs
+++ b/crates/wasi/src/tcp.rs
@@ -3,10 +3,10 @@ use crate::host::network;
 use crate::network::SocketAddressFamily;
 use crate::runtime::{with_ambient_tokio_runtime, AbortOnDropJoinHandle};
 use crate::{
-    HostInputStream, HostOutputStream, InputStream, OutputStream, SocketResult, StreamError,
-    Subscribe,
+    HostInputStream, HostOutputStream, InputStream, OutputStream, SocketError, SocketResult,
+    StreamError, Subscribe,
 };
-use anyhow::{Error, Result};
+use anyhow::Result;
 use cap_net_ext::AddressFamily;
 use futures::Future;
 use io_lifetimes::views::SocketlikeView;
@@ -15,10 +15,11 @@ use rustix::io::Errno;
 use rustix::net::sockopt;
 use std::io;
 use std::mem;
-use std::net::SocketAddr;
+use std::net::{Shutdown, SocketAddr};
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::Poll;
+use tokio::sync::Mutex;
 
 /// Value taken from rust std library.
 const DEFAULT_BACKLOG: u32 = 128;
@@ -54,7 +55,13 @@ enum TcpState {
     ConnectReady(io::Result<tokio::net::TcpStream>),
 
     /// An outgoing connection has been established.
-    Connected(Arc<tokio::net::TcpStream>),
+    Connected {
+        stream: Arc<tokio::net::TcpStream>,
+
+        // WASI is single threaded, so in practice these Mutexes should never be contended:
+        reader: Arc<Mutex<TcpReader>>,
+        writer: Arc<Mutex<TcpWriter>>,
+    },
 
     Closed,
 }
@@ -72,7 +79,7 @@ impl std::fmt::Debug for TcpState {
                 .finish(),
             Self::Connecting(_) => f.debug_tuple("Connecting").finish(),
             Self::ConnectReady(_) => f.debug_tuple("ConnectReady").finish(),
-            Self::Connected(_) => f.debug_tuple("Connected").finish(),
+            Self::Connected { .. } => f.debug_tuple("Connected").finish(),
             Self::Closed => write!(f, "Closed"),
         }
     }
@@ -145,7 +152,9 @@ impl TcpSocket {
             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::Connected { stream, .. } => {
+                Ok(stream.as_socketlike_view::<std::net::TcpStream>())
+            }
             TcpState::Listening { listener, .. } => {
                 Ok(listener.as_socketlike_view::<std::net::TcpStream>())
             }
@@ -277,9 +286,15 @@ impl TcpSocket {
         match result {
             Ok(stream) => {
                 let stream = Arc::new(stream);
-                self.tcp_state = TcpState::Connected(stream.clone());
-                let input: InputStream = Box::new(TcpReadStream::new(stream.clone()));
-                let output: OutputStream = Box::new(TcpWriteStream::new(stream));
+                let reader = Arc::new(Mutex::new(TcpReader::new(stream.clone())));
+                let writer = Arc::new(Mutex::new(TcpWriter::new(stream.clone())));
+                self.tcp_state = TcpState::Connected {
+                    stream,
+                    reader: reader.clone(),
+                    writer: writer.clone(),
+                };
+                let input: InputStream = Box::new(TcpReadStream(reader));
+                let output: OutputStream = Box::new(TcpWriteStream(writer));
                 Ok((input, output))
             }
             Err(err) => {
@@ -427,9 +442,19 @@ impl TcpSocket {
 
         let client = Arc::new(client);
 
-        let input: InputStream = 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)?;
+        let reader = Arc::new(Mutex::new(TcpReader::new(client.clone())));
+        let writer = Arc::new(Mutex::new(TcpWriter::new(client.clone())));
+
+        let input: InputStream = Box::new(TcpReadStream(reader.clone()));
+        let output: OutputStream = Box::new(TcpWriteStream(writer.clone()));
+        let tcp_socket = TcpSocket::from_state(
+            TcpState::Connected {
+                stream: client,
+                reader,
+                writer,
+            },
+            self.family,
+        )?;
 
         Ok((tcp_socket, input, output))
     }
@@ -446,7 +471,7 @@ impl TcpSocket {
 
     pub fn remote_address(&self) -> SocketResult<SocketAddr> {
         let view = match self.tcp_state {
-            TcpState::Connected(..) => self.as_std_view()?,
+            TcpState::Connected { .. } => self.as_std_view()?,
             TcpState::Connecting(..) | TcpState::ConnectReady(..) => {
                 return Err(ErrorCode::ConcurrencyConflict.into())
             }
@@ -613,20 +638,19 @@ impl TcpSocket {
         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",
-                ))
-            }
+    pub fn shutdown(&self, how: Shutdown) -> SocketResult<()> {
+        let TcpState::Connected { reader, writer, .. } = &self.tcp_state else {
+            return Err(ErrorCode::InvalidState.into());
         };
 
-        stream
-            .as_socketlike_view::<std::net::TcpStream>()
-            .shutdown(how)?;
+        if let Shutdown::Both | Shutdown::Read = how {
+            try_lock_for_socket(reader)?.shutdown();
+        }
+
+        if let Shutdown::Both | Shutdown::Write = how {
+            try_lock_for_socket(writer)?.shutdown();
+        }
+
         Ok(())
     }
 }
@@ -641,7 +665,7 @@ impl Subscribe for TcpSocket {
             | TcpState::ListenStarted(..)
             | TcpState::ConnectReady(..)
             | TcpState::Closed
-            | TcpState::Connected(..) => {
+            | TcpState::Connected { .. } => {
                 // No async operation in progress.
             }
             TcpState::Connecting(future) => {
@@ -664,22 +688,18 @@ impl Subscribe for TcpSocket {
     }
 }
 
-struct TcpReadStream {
+struct TcpReader {
     stream: Arc<tokio::net::TcpStream>,
     closed: bool,
 }
 
-impl TcpReadStream {
+impl TcpReader {
     fn new(stream: Arc<tokio::net::TcpStream>) -> Self {
         Self {
             stream,
             closed: false,
         }
     }
-}
-
-#[async_trait::async_trait]
-impl HostInputStream for TcpReadStream {
     fn read(&mut self, size: usize) -> Result<bytes::Bytes, StreamError> {
         if self.closed {
             return Err(StreamError::Closed);
@@ -710,37 +730,57 @@ impl HostInputStream for TcpReadStream {
         buf.truncate(n);
         Ok(buf.freeze())
     }
-}
 
-#[async_trait::async_trait]
-impl Subscribe for TcpReadStream {
+    fn shutdown(&mut self) {
+        native_shutdown(&self.stream, Shutdown::Read);
+        self.closed = true;
+    }
+
     async fn ready(&mut self) {
         if self.closed {
             return;
         }
+
         self.stream.readable().await.unwrap();
     }
 }
 
+struct TcpReadStream(Arc<Mutex<TcpReader>>);
+
+#[async_trait::async_trait]
+impl HostInputStream for TcpReadStream {
+    fn read(&mut self, size: usize) -> Result<bytes::Bytes, StreamError> {
+        try_lock_for_stream(&self.0)?.read(size)
+    }
+}
+
+#[async_trait::async_trait]
+impl Subscribe for TcpReadStream {
+    async fn ready(&mut self) {
+        self.0.lock().await.ready().await
+    }
+}
+
 const SOCKET_READY_SIZE: usize = 1024 * 1024 * 1024;
 
-struct TcpWriteStream {
+struct TcpWriter {
     stream: Arc<tokio::net::TcpStream>,
-    last_write: LastWrite,
+    state: WriteState,
 }
 
-enum LastWrite {
-    Waiting(AbortOnDropJoinHandle<Result<()>>),
-    Error(Error),
-    Done,
+enum WriteState {
+    Ready,
+    Writing(AbortOnDropJoinHandle<io::Result<()>>),
+    Closing(AbortOnDropJoinHandle<io::Result<()>>),
     Closed,
+    Error(io::Error),
 }
 
-impl TcpWriteStream {
+impl TcpWriter {
     fn new(stream: Arc<tokio::net::TcpStream>) -> Self {
         Self {
             stream,
-            last_write: LastWrite::Done,
+            state: WriteState::Ready,
         }
     }
 
@@ -761,10 +801,10 @@ impl TcpWriteStream {
     /// Write `bytes` in a background task, remembering the task handle for use in a future call to
     /// `write_ready`
     fn background_write(&mut self, mut bytes: bytes::Bytes) {
-        assert!(matches!(self.last_write, LastWrite::Done));
+        assert!(matches!(self.state, WriteState::Ready));
 
         let stream = self.stream.clone();
-        self.last_write = LastWrite::Waiting(crate::runtime::spawn(async move {
+        self.state = WriteState::Writing(crate::runtime::spawn(async move {
             // Note: we are not using the AsyncWrite impl here, and instead using the TcpStream
             // primitive try_write, which goes directly to attempt a write with mio. This has
             // two advantages: 1. this operation takes a &TcpStream instead of a &mut TcpStream
@@ -784,14 +824,12 @@ impl TcpWriteStream {
             Ok(())
         }));
     }
-}
 
-#[async_trait::async_trait]
-impl HostOutputStream for TcpWriteStream {
     fn write(&mut self, mut bytes: bytes::Bytes) -> Result<(), StreamError> {
-        match self.last_write {
-            LastWrite::Done => {}
-            LastWrite::Waiting(_) | LastWrite::Error(_) | LastWrite::Closed => {
+        match self.state {
+            WriteState::Ready => {}
+            WriteState::Closed => return Err(StreamError::Closed),
+            WriteState::Writing(_) | WriteState::Closing(_) | WriteState::Error(_) => {
                 return Err(StreamError::Trap(anyhow::anyhow!(
                     "unpermitted: must call check_write first"
                 )));
@@ -812,7 +850,7 @@ impl HostOutputStream for TcpWriteStream {
                 }
 
                 Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => {
-                    self.last_write = LastWrite::Closed;
+                    self.state = WriteState::Closed;
                     return Err(StreamError::Closed);
                 }
 
@@ -827,23 +865,30 @@ impl HostOutputStream for TcpWriteStream {
         // `flush` is a no-op here, as we're not managing any internal buffer. Additionally,
         // `write_ready` will join the background write task if it's active, so following `flush`
         // with `write_ready` will have the desired effect.
-        match self.last_write {
-            LastWrite::Done | LastWrite::Waiting(_) | LastWrite::Error(_) => Ok(()),
-            LastWrite::Closed => Err(StreamError::Closed),
+        match self.state {
+            WriteState::Ready
+            | WriteState::Writing(_)
+            | WriteState::Closing(_)
+            | WriteState::Error(_) => Ok(()),
+            WriteState::Closed => Err(StreamError::Closed),
         }
     }
 
     fn check_write(&mut self) -> Result<usize, StreamError> {
-        match mem::replace(&mut self.last_write, LastWrite::Closed) {
-            LastWrite::Waiting(task) => {
-                self.last_write = LastWrite::Waiting(task);
+        match mem::replace(&mut self.state, WriteState::Closed) {
+            WriteState::Writing(task) => {
+                self.state = WriteState::Writing(task);
                 return Ok(0);
             }
-            LastWrite::Done => {
-                self.last_write = LastWrite::Done;
+            WriteState::Closing(task) => {
+                self.state = WriteState::Closing(task);
+                return Ok(0);
+            }
+            WriteState::Ready => {
+                self.state = WriteState::Ready;
             }
-            LastWrite::Closed => return Err(StreamError::Closed),
-            LastWrite::Error(e) => return Err(StreamError::LastOperationFailed(e.into())),
+            WriteState::Closed => return Err(StreamError::Closed),
+            WriteState::Error(e) => return Err(StreamError::LastOperationFailed(e.into())),
         }
 
         let writable = self.stream.writable();
@@ -853,25 +898,103 @@ impl HostOutputStream for TcpWriteStream {
         }
         Ok(SOCKET_READY_SIZE)
     }
+
+    fn shutdown(&mut self) {
+        self.state = match mem::replace(&mut self.state, WriteState::Closed) {
+            // No write in progress, immediately shut down:
+            WriteState::Ready => {
+                native_shutdown(&self.stream, Shutdown::Write);
+                WriteState::Closed
+            }
+
+            // Schedule the shutdown after the current write has finished:
+            WriteState::Writing(write) => {
+                let stream = self.stream.clone();
+                WriteState::Closing(crate::runtime::spawn(async move {
+                    let result = write.await;
+                    native_shutdown(&stream, Shutdown::Write);
+                    result
+                }))
+            }
+
+            s => s,
+        };
+    }
+
     async fn cancel(&mut self) {
-        match mem::replace(&mut self.last_write, LastWrite::Closed) {
-            LastWrite::Waiting(task) => _ = task.abort_wait().await,
+        match mem::replace(&mut self.state, WriteState::Closed) {
+            WriteState::Writing(task) | WriteState::Closing(task) => _ = task.cancel().await,
             _ => {}
         }
     }
-}
 
-#[async_trait::async_trait]
-impl Subscribe for TcpWriteStream {
     async fn ready(&mut self) {
-        if let LastWrite::Waiting(task) = &mut self.last_write {
-            self.last_write = match task.await {
-                Ok(()) => LastWrite::Done,
-                Err(e) => LastWrite::Error(e),
-            };
+        match &mut self.state {
+            WriteState::Writing(task) => {
+                self.state = match task.await {
+                    Ok(()) => WriteState::Ready,
+                    Err(e) => WriteState::Error(e),
+                }
+            }
+            WriteState::Closing(task) => {
+                self.state = match task.await {
+                    Ok(()) => WriteState::Closed,
+                    Err(e) => WriteState::Error(e),
+                }
+            }
+            _ => {}
         }
-        if let LastWrite::Done = self.last_write {
+
+        if let WriteState::Ready = self.state {
             self.stream.writable().await.unwrap();
         }
     }
 }
+
+struct TcpWriteStream(Arc<Mutex<TcpWriter>>);
+
+#[async_trait::async_trait]
+impl HostOutputStream for TcpWriteStream {
+    fn write(&mut self, bytes: bytes::Bytes) -> Result<(), StreamError> {
+        try_lock_for_stream(&self.0)?.write(bytes)
+    }
+
+    fn flush(&mut self) -> Result<(), StreamError> {
+        try_lock_for_stream(&self.0)?.flush()
+    }
+
+    fn check_write(&mut self) -> Result<usize, StreamError> {
+        try_lock_for_stream(&self.0)?.check_write()
+    }
+
+    async fn cancel(&mut self) {
+        self.0.lock().await.cancel().await
+    }
+}
+
+#[async_trait::async_trait]
+impl Subscribe for TcpWriteStream {
+    async fn ready(&mut self) {
+        self.0.lock().await.ready().await
+    }
+}
+
+fn native_shutdown(stream: &tokio::net::TcpStream, how: Shutdown) {
+    _ = stream
+        .as_socketlike_view::<std::net::TcpStream>()
+        .shutdown(how);
+}
+
+fn try_lock_for_stream<T>(mutex: &Mutex<T>) -> Result<tokio::sync::MutexGuard<'_, T>, StreamError> {
+    mutex
+        .try_lock()
+        .map_err(|_| StreamError::trap("concurrent access to resource not supported"))
+}
+
+fn try_lock_for_socket<T>(mutex: &Mutex<T>) -> Result<tokio::sync::MutexGuard<'_, T>, SocketError> {
+    mutex.try_lock().map_err(|_| {
+        SocketError::trap(anyhow::anyhow!(
+            "concurrent access to resource not supported"
+        ))
+    })
+}
diff --git a/crates/wasi/src/write_stream.rs b/crates/wasi/src/write_stream.rs
index 33ae91048..fe3658662 100644
--- a/crates/wasi/src/write_stream.rs
+++ b/crates/wasi/src/write_stream.rs
@@ -198,7 +198,7 @@ impl HostOutputStream for AsyncWriteStream {
 
     async fn cancel(&mut self) {
         match self.join_handle.take() {
-            Some(task) => _ = task.abort_wait().await,
+            Some(task) => _ = task.cancel().await,
             None => {}
         }
     }
diff --git a/crates/wasmtime/src/runtime/vm/sys/custom/mmap.rs b/crates/wasmtime/src/runtime/vm/sys/custom/mmap.rs
index a9f462627..171cf010a 100644
--- a/crates/wasmtime/src/runtime/vm/sys/custom/mmap.rs
+++ b/crates/wasmtime/src/runtime/vm/sys/custom/mmap.rs
@@ -67,7 +67,7 @@ impl Mmap {
 
     #[inline]
     pub fn len(&self) -> usize {
-        self.memory.as_ptr().len()
+        unsafe { (*self.memory.as_ptr()).len() }
     }
 
     pub unsafe fn make_executable(
@@ -102,7 +102,7 @@ impl Drop for Mmap {
     fn drop(&mut self) {
         unsafe {
             let ptr = self.memory.as_ptr().cast();
-            let len = self.memory.as_ptr().len();
+            let len = (*self.memory.as_ptr()).len();
             if len == 0 {
                 return;
             }
diff --git a/crates/wasmtime/src/runtime/vm/sys/miri/mmap.rs b/crates/wasmtime/src/runtime/vm/sys/miri/mmap.rs
index b4f834f3a..de2a6aa3f 100644
--- a/crates/wasmtime/src/runtime/vm/sys/miri/mmap.rs
+++ b/crates/wasmtime/src/runtime/vm/sys/miri/mmap.rs
@@ -68,7 +68,7 @@ impl Mmap {
     }
 
     pub fn len(&self) -> usize {
-        self.memory.as_ptr().len()
+        unsafe { (*self.memory.as_ptr()).len() }
     }
 
     pub unsafe fn make_executable(
diff --git a/crates/wasmtime/src/runtime/vm/sys/unix/mmap.rs b/crates/wasmtime/src/runtime/vm/sys/unix/mmap.rs
index ad6f230e1..4c29a9ae1 100644
--- a/crates/wasmtime/src/runtime/vm/sys/unix/mmap.rs
+++ b/crates/wasmtime/src/runtime/vm/sys/unix/mmap.rs
@@ -104,7 +104,7 @@ impl Mmap {
 
     #[inline]
     pub fn len(&self) -> usize {
-        self.memory.as_ptr().len()
+        unsafe { (*self.memory.as_ptr()).len() }
     }
 
     pub unsafe fn make_executable(
@@ -149,7 +149,7 @@ impl Drop for Mmap {
     fn drop(&mut self) {
         unsafe {
             let ptr = self.memory.as_ptr().cast();
-            let len = self.memory.as_ptr().len();
+            let len = (*self.memory.as_ptr()).len();
             if len == 0 {
                 return;
             }
diff --git a/crates/wasmtime/src/runtime/vm/sys/windows/mmap.rs b/crates/wasmtime/src/runtime/vm/sys/windows/mmap.rs
index c0e6bc0ef..d4d62061e 100644
--- a/crates/wasmtime/src/runtime/vm/sys/windows/mmap.rs
+++ b/crates/wasmtime/src/runtime/vm/sys/windows/mmap.rs
@@ -164,7 +164,7 @@ impl Mmap {
 
     #[inline]
     pub fn len(&self) -> usize {
-        self.memory.as_ptr().len()
+        unsafe { (*self.memory.as_ptr()).len() }
     }
 
     pub unsafe fn make_executable(


Test patch:
diff --git a/crates/test-programs/src/bin/preview2_tcp_streams.rs b/crates/test-programs/src/bin/preview2_tcp_streams.rs
index 688827be6..c75695896 100644
--- a/crates/test-programs/src/bin/preview2_tcp_streams.rs
+++ b/crates/test-programs/src/bin/preview2_tcp_streams.rs
@@ -25,6 +25,31 @@ fn test_tcp_input_stream_should_be_closed_by_remote_shutdown(
     });
 }
 
+/// InputStream::read should return `StreamError::Closed` after the connection has been shut down locally.
+fn test_tcp_input_stream_should_be_closed_by_local_shutdown(
+    net: &Network,
+    family: IpAddressFamily,
+) {
+    setup(net, family, |server, client| {
+        // On Linux, `recv` continues to work even after `shutdown(sock, SHUT_RD)`
+        // has been called. To properly test that this behavior doesn't happen in
+        // WASI, we make sure there's some data to read by the client:
+        server.output.blocking_write_util(b"Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.").unwrap();
+
+        // Wait for the data to reach the client:
+        client.input.subscribe().block();
+
+        // Shut down socket locally:
+        client.socket.shutdown(ShutdownType::Receive).unwrap();
+
+        // The input stream should immediately signal StreamError::Closed.
+        assert!(matches!(client.input.read(10), Err(StreamError::Closed)));
+
+        // Stream should still be closed, even when requesting 0 bytes:
+        assert!(matches!(client.input.read(0), Err(StreamError::Closed)));
+    });
+}
+
 /// OutputStream should return `StreamError::Closed` after the connection has been locally shut down for sending.
 fn test_tcp_output_stream_should_be_closed_by_local_shutdown(
     net: &Network,
@@ -54,14 +79,50 @@ fn test_tcp_output_stream_should_be_closed_by_local_shutdown(
     });
 }
 
+/// Calling `shutdown` while the OutputStream is in the middle of a background write should not cause that write to be lost.
+fn test_tcp_shutdown_should_not_lose_data(net: &Network, family: IpAddressFamily) {
+    setup(net, family, |server, client| {
+        // Minimize the local send buffer:
+        client.socket.set_send_buffer_size(1024).unwrap();
+        let small_buffer_size = client.socket.send_buffer_size().unwrap();
+
+        // Create a significantly bigger buffer, so that we can be pretty sure the `write` won't finish immediately:
+        let big_buffer_size = client
+            .output
+            .check_write()
+            .unwrap()
+            .min(100 * small_buffer_size);
+        assert!(big_buffer_size > small_buffer_size);
+        let outgoing_data = vec![0; big_buffer_size as usize];
+
+        // Submit the oversized buffer and immediately initiate the shutdown:
+        client.output.write(&outgoing_data).unwrap();
+        client.socket.shutdown(ShutdownType::Send).unwrap();
+
+        // The peer should receive _all_ data:
+        let incoming_data = server.input.blocking_read_to_end().unwrap();
+        assert_eq!(
+            outgoing_data.len(),
+            incoming_data.len(),
+            "Received data should match the sent data"
+        );
+    });
+}
+
 fn main() {
     let net = Network::default();
 
     test_tcp_input_stream_should_be_closed_by_remote_shutdown(&net, IpAddressFamily::Ipv4);
     test_tcp_input_stream_should_be_closed_by_remote_shutdown(&net, IpAddressFamily::Ipv6);
 
+    test_tcp_input_stream_should_be_closed_by_local_shutdown(&net, IpAddressFamily::Ipv4);
+    test_tcp_input_stream_should_be_closed_by_local_shutdown(&net, IpAddressFamily::Ipv6);
+
     test_tcp_output_stream_should_be_closed_by_local_shutdown(&net, IpAddressFamily::Ipv4);
     test_tcp_output_stream_should_be_closed_by_local_shutdown(&net, IpAddressFamily::Ipv6);
+
+    test_tcp_shutdown_should_not_lose_data(&net, IpAddressFamily::Ipv4);
+    test_tcp_shutdown_should_not_lose_data(&net, IpAddressFamily::Ipv6);
 }
 
 struct Connection {
diff --git a/crates/test-programs/src/sockets.rs b/crates/test-programs/src/sockets.rs
index c77ad8d7f..0fcccaaab 100644
--- a/crates/test-programs/src/sockets.rs
+++ b/crates/test-programs/src/sockets.rs
@@ -29,6 +29,19 @@ impl Pollable {
     }
 }
 
+impl InputStream {
+    pub fn blocking_read_to_end(&self) -> Result<Vec<u8>, crate::wasi::io::error::Error> {
+        let mut data = vec![];
+        loop {
+            match self.blocking_read(1024 * 1024) {
+                Ok(chunk) => data.extend(chunk),
+                Err(StreamError::Closed) => return Ok(data),
+                Err(StreamError::LastOperationFailed(e)) => return Err(e),
+            }
+        }
+    }
+}
+
 impl OutputStream {
     pub fn blocking_write_util(&self, mut bytes: &[u8]) -> Result<(), StreamError> {
         let timeout = monotonic_clock::subscribe_duration(TIMEOUT_NS);