Wasmtime is vulnerable to panic when dropping a `[Typed]Func::call_async` future

@@ -492,6 +492,18 @@ wasmtime_option_group! {
         pub keyvalue_in_memory_data: Vec<KeyValuePair>,
         /// Enable support for WASIp3 APIs.
         pub p3: Option<bool>,
+        /// Maximum resources the guest is allowed to create simultaneously.
+        pub max_resources: Option<usize>,
+        /// Fuel to use for all hostcalls to limit guest<->host data transfer.
+        pub hostcall_fuel: Option<usize>,
+        /// Maximum value, in bytes, for a wasi-random 0.2
+        /// `get{,-insecure}-random-bytes` `len` parameter. Calls with a value
+        /// exceeding this limit will trap.
+        pub max_random_size: Option<u64>,
+        /// Maximum value, in bytes, for the contents of a wasi-http 0.2
+        /// `fields` resource (aka `headers` and `trailers`). `fields` methods
+        /// which cause the contents to exceed this size limit will trap.
+        pub max_http_fields_size: Option<usize>,
     }
 
     enum Wasi {

@@ -73,21 +73,26 @@ impl Artifacts {
             // Bucket, based on the name of the test, into a "kind" which
             // generates a `foreach_*` macro below.
             let kind = match test.name.as_str() {
+                s if s.starts_with("p1_cli_")
+                    || s.starts_with("p2_cli_")
+                    || s.starts_with("p3_cli_") =>
+                {
+                    "cli"
+                }
                 s if s.starts_with("p1_") => "p1",
                 s if s.starts_with("p2_http_") => "p2_http",
-                s if s.starts_with("p2_cli_") => "p2_cli",
                 s if s.starts_with("p2_api_") => "p2_api",
                 s if s.starts_with("p2_") => "p2",
+                s if s.starts_with("p3_http_") => "p3_http",
+                s if s.starts_with("p3_api_") => "p3_api",
+                s if s.starts_with("p3_") => "p3",
                 s if s.starts_with("nn_") => "nn",
                 s if s.starts_with("piped_") => "piped",
                 s if s.starts_with("dwarf_") => "dwarf",
                 s if s.starts_with("config_") => "config",
                 s if s.starts_with("keyvalue_") => "keyvalue",
                 s if s.starts_with("tls_") => "tls",
                 s if s.starts_with("async_") => "async",
-                s if s.starts_with("p3_http_") => "p3_http",
-                s if s.starts_with("p3_api_") => "p3_api",
-                s if s.starts_with("p3_") => "p3",
                 // If you're reading this because you hit this panic, either add
                 // it to a test suite above or add a new "suite". The purpose of
                 // the categorization above is to have a static assertion that
@@ -271,7 +276,7 @@ impl Artifacts {
                 // Prevent stray files for now that we don't understand.
                 Some(_) => panic!("unknown file extension on {path:?}"),
 
-                None => unreachable!(),
+                None => unreachable!("no extension in path {path:?}"),
             }
         }
     }

@@ -0,0 +1,118 @@
+use std::ptr;
+
+fn main() {
+    big_poll();
+    big_string();
+    big_iovecs();
+}
+
+fn big_string() {
+    let mut s = String::new();
+    for _ in 0..10_000 {
+        s.push_str("hello world");
+    }
+    let dir_fd = test_programs::preview1::open_scratch_directory(".").unwrap();
+    assert_eq!(
+        unsafe { wasip1::path_create_directory(dir_fd, &s) },
+        Err(wasip1::ERRNO_NOMEM)
+    );
+}
+
+fn big_iovecs() {
+    let mut iovs = Vec::new();
+    let mut ciovs = Vec::new();
+    for _ in 0..10_000 {
+        iovs.push(wasip1::Iovec {
+            buf: ptr::null_mut(),
+            buf_len: 0,
+        });
+        ciovs.push(wasip1::Ciovec {
+            buf: ptr::null(),
+            buf_len: 0,
+        });
+    }
+    let dir_fd = test_programs::preview1::open_scratch_directory(".").unwrap();
+    let fd = unsafe {
+        wasip1::path_open(
+            dir_fd,
+            0,
+            "hi",
+            wasip1::OFLAGS_CREAT,
+            wasip1::RIGHTS_FD_WRITE | wasip1::RIGHTS_FD_READ,
+            0,
+            0,
+        )
+        .unwrap()
+    };
+
+    unsafe {
+        assert_eq!(wasip1::fd_write(fd, &ciovs), Err(wasip1::ERRNO_NOMEM));
+        assert_eq!(wasip1::fd_read(fd, &iovs), Err(wasip1::ERRNO_NOMEM));
+        assert_eq!(wasip1::fd_pwrite(fd, &ciovs, 0), Err(wasip1::ERRNO_NOMEM));
+        assert_eq!(wasip1::fd_pread(fd, &iovs, 0), Err(wasip1::ERRNO_NOMEM));
+    }
+
+    ciovs.truncate(1);
+    iovs.truncate(1);
+    iovs.push(wasip1::Iovec {
+        buf: ptr::null_mut(),
+        buf_len: 10_000,
+    });
+    ciovs.push(wasip1::Ciovec {
+        buf: ptr::null(),
+        buf_len: 10_000,
+    });
+    unsafe {
+        assert_eq!(wasip1::fd_write(fd, &ciovs), Err(wasip1::ERRNO_NOMEM));
+        assert_eq!(wasip1::fd_read(fd, &iovs), Err(wasip1::ERRNO_NOMEM));
+        assert_eq!(wasip1::fd_pwrite(fd, &ciovs, 0), Err(wasip1::ERRNO_NOMEM));
+        assert_eq!(wasip1::fd_pread(fd, &iovs, 0), Err(wasip1::ERRNO_NOMEM));
+    }
+}
+
+fn big_poll() {
+    let mut huge_poll = Vec::new();
+    let mut huge_events = Vec::new();
+    for _ in 0..10_000 {
+        huge_poll.push(subscribe_timeout(0));
+        huge_events.push(empty_event());
+    }
+    let err = unsafe {
+        wasip1::poll_oneoff(
+            huge_poll.as_ptr(),
+            huge_events.as_mut_ptr(),
+            huge_poll.len(),
+        )
+        .unwrap_err()
+    };
+    assert_eq!(err, wasip1::ERRNO_NOMEM);
+
+    fn subscribe_timeout(timeout: u64) -> wasip1::Subscription {
+        wasip1::Subscription {
+            userdata: 0,
+            u: wasip1::SubscriptionU {
+                tag: wasip1::EVENTTYPE_CLOCK.raw(),
+                u: wasip1::SubscriptionUU {
+                    clock: wasip1::SubscriptionClock {
+                        id: wasip1::CLOCKID_MONOTONIC,
+                        timeout,
+                        precision: 0,
+                        flags: 0,
+                    },
+                },
+            },
+        }
+    }
+
+    fn empty_event() -> wasip1::Event {
+        wasip1::Event {
+            error: wasip1::ERRNO_SUCCESS,
+            fd_readwrite: wasip1::EventFdReadwrite {
+                nbytes: 0,
+                flags: 0,
+            },
+            type_: wasip1::EVENTTYPE_CLOCK,
+            userdata: 0,
+        }
+    }
+}

@@ -0,0 +1,98 @@
+use std::process;
+use test_programs::preview1::open_scratch_directory;
+
+fn main() {
+    let arg = std::env::args().nth(1).unwrap();
+    let dir_fd = match open_scratch_directory(&arg) {
+        Ok(dir_fd) => dir_fd,
+        Err(err) => {
+            eprintln!("{err}");
+            process::exit(1)
+        }
+    };
+
+    // Wait for just one timeout (maybe hitting a fast path)
+    let subs = [subscribe_timeout(0)];
+    for _ in 0..1000 {
+        let mut events = [empty_event()];
+        unsafe {
+            wasip1::poll_oneoff(subs.as_ptr(), events.as_mut_ptr(), 1).unwrap();
+        }
+    }
+
+    // Wait for two timeouts
+    let subs = [subscribe_timeout(0), subscribe_timeout(0)];
+    for _ in 0..1000 {
+        let mut events = [empty_event(), empty_event()];
+        unsafe {
+            wasip1::poll_oneoff(subs.as_ptr(), events.as_mut_ptr(), 2).unwrap();
+        }
+    }
+
+    let file_fd = unsafe {
+        wasip1::path_open(
+            dir_fd,
+            0,
+            "hello.txt",
+            wasip1::OFLAGS_CREAT,
+            wasip1::RIGHTS_FD_WRITE | wasip1::RIGHTS_FD_READ,
+            0,
+            0,
+        )
+        .expect("creating a file for writing")
+    };
+
+    // Wait for a timeout fd operations
+    let subs = [
+        subscribe_timeout(0),
+        subscribe_fd(wasip1::EVENTTYPE_FD_READ, file_fd),
+        subscribe_fd(wasip1::EVENTTYPE_FD_WRITE, file_fd),
+    ];
+    for _ in 0..1000 {
+        let mut events = [empty_event(), empty_event(), empty_event()];
+        unsafe {
+            wasip1::poll_oneoff(subs.as_ptr(), events.as_mut_ptr(), 3).unwrap();
+        }
+    }
+}
+
+fn subscribe_timeout(timeout: u64) -> wasip1::Subscription {
+    wasip1::Subscription {
+        userdata: 0,
+        u: wasip1::SubscriptionU {
+            tag: wasip1::EVENTTYPE_CLOCK.raw(),
+            u: wasip1::SubscriptionUU {
+                clock: wasip1::SubscriptionClock {
+                    id: wasip1::CLOCKID_MONOTONIC,
+                    timeout,
+                    precision: 0,
+                    flags: 0,
+                },
+            },
+        },
+    }
+}
+
+fn subscribe_fd(ty: wasip1::Eventtype, file_descriptor: wasip1::Fd) -> wasip1::Subscription {
+    wasip1::Subscription {
+        userdata: 0,
+        u: wasip1::SubscriptionU {
+            tag: ty.raw(),
+            u: wasip1::SubscriptionUU {
+                fd_read: wasip1::SubscriptionFdReadwrite { file_descriptor },
+            },
+        },
+    }
+}
+
+fn empty_event() -> wasip1::Event {
+    wasip1::Event {
+        error: wasip1::ERRNO_SUCCESS,
+        fd_readwrite: wasip1::EventFdReadwrite {
+            nbytes: 0,
+            flags: 0,
+        },
+        type_: wasip1::EVENTTYPE_CLOCK,
+        userdata: 0,
+    }
+}

@@ -1,3 +1,4 @@
+use anyhow::{Context, Result};
 use test_programs::wasi::http::types::{
     Headers, IncomingRequest, Method, OutgoingBody, OutgoingResponse, ResponseOutparam,
 };
@@ -29,6 +30,16 @@ impl test_programs::proxy::exports::wasi::http::incoming_handler::Guest for T {
 
                 return;
             }
+            (Method::Get, Some(p)) if p.starts_with("/modify_fields/") => {
+                let r = modify_fields_handler(request);
+                response_for(r, outparam);
+                return;
+            }
+            (Method::Get, Some(p)) if p.starts_with("/new_fields/") => {
+                let r = new_fields_handler(request);
+                response_for(r, outparam);
+                return;
+            }
 
             _ => {}
         }
@@ -69,10 +80,64 @@ impl test_programs::proxy::exports::wasi::http::incoming_handler::Guest for T {
     }
 }
 
+fn response_for(r: Result<()>, outparam: ResponseOutparam) {
+    let resp = OutgoingResponse::new(Headers::new());
+    resp.set_status_code(if r.is_ok() { 200 } else { 500 })
+        .unwrap();
+    let body = resp.body().expect("outgoing response");
+    ResponseOutparam::set(outparam, Ok(resp));
+    let _ = body.write().and_then(|out| {
+        let _ = out.blocking_write_and_flush(format!("{r:?}").as_bytes());
+        drop(out);
+        Ok(())
+    });
+    let _ = OutgoingBody::finish(body, None);
+}
+
 // Technically this should not be here for a proxy, but given the current
 // framework for tests it's required since this file is built as a `bin`
 fn main() {}
 
 fn test_filesystem() {
     assert!(std::fs::File::open(".").is_err());
 }
+
+fn add_bytes_to_headers(headers: Headers, size: usize) {
+    if size == 0 {
+        return;
+    } else if size < 10 {
+        headers.append("k", &b"abcdefghi"[0..size - 1]).unwrap()
+    } else {
+        for chunk in 0..(size / 10) {
+            let k = format!("g{chunk:04}");
+            let mut v = format!("h{chunk:04}");
+            if chunk == 0 {
+                for _ in 0..(size % 10) {
+                    v.push('#');
+                }
+            }
+            headers.append(k.as_str(), v.as_bytes()).unwrap()
+        }
+    }
+}
+
+fn modify_fields_handler(request: IncomingRequest) -> Result<()> {
+    let path = request.path_with_query().unwrap();
+    let rest = path.trim_start_matches("/modify_fields/");
+    let added_field_bytes: usize = rest
+        .parse()
+        .context("expect remainder of url to parse as number")?;
+    add_bytes_to_headers(request.headers().clone(), added_field_bytes);
+
+    Ok(())
+}
+fn new_fields_handler(request: IncomingRequest) -> Result<()> {
+    let path = request.path_with_query().unwrap();
+    let rest = path.trim_start_matches("/new_fields/");
+    let added_field_bytes: usize = rest
+        .parse()
+        .context("expect remainder of url to parse as number")?;
+    add_bytes_to_headers(Headers::new(), added_field_bytes);
+
+    Ok(())
+}

@@ -0,0 +1,282 @@
+use wasip2::filesystem::types::*;
+use wasip2::sockets::network::*;
+
+fn main() {
+    match std::env::args().nth(1).as_deref() {
+        Some("poll") => poll(),
+        Some("read") => read(),
+        Some("write") => write(),
+        Some("write-stream") => write_stream(),
+        Some("write-stream-blocking") => write_stream_blocking(),
+        Some("mkdir") => mkdir(),
+        Some("resolve") => resolve(),
+        Some("udp-send-many") => udp_send_many(),
+        Some("udp-send-big") => udp_send_big(),
+        Some("write-zeroes") => write_zeroes(),
+        Some("write-stream-buffer-too-large") => write_stream_buffer_too_large(),
+        Some("write-zeroes-buffer-too-large") => write_zeroes_buffer_too_large(),
+        Some("read-file-big") => read_file_big(),
+        Some("read-tcp-big") => read_tcp_big(),
+        other => panic!("unknown arg {other:?}"),
+    }
+}
+
+fn poll() {
+    let mut events = Vec::new();
+    let sub = wasip2::clocks::monotonic_clock::subscribe_duration(0);
+    for _ in 0..5000 {
+        events.push(&sub);
+    }
+
+    wasip2::io::poll::poll(&events);
+    unreachable!()
+}
+
+fn preopen() -> Descriptor {
+    let mut dirs = wasip2::filesystem::preopens::get_directories();
+    assert_eq!(dirs.len(), 1);
+    dirs.pop().unwrap().0
+}
+
+fn read() {
+    let f = preopen()
+        .open_at(
+            PathFlags::empty(),
+            "1mb",
+            OpenFlags::empty(),
+            DescriptorFlags::empty(),
+        )
+        .unwrap();
+
+    // 0-length is ok
+    f.read(0, 0).unwrap();
+
+    // This isn't transferring data from the guest to the host, so this is ok.
+    f.read(1 << 20, 0).unwrap();
+
+    // Host shouldn't allocate based on what the guest asks for...
+    f.read(u64::MAX, 0).unwrap();
+
+    f.read_via_stream(0)
+        .unwrap()
+        .blocking_read(1 << 20)
+        .unwrap();
+}
+
+fn write() {
+    let f = preopen()
+        .open_at(
+            PathFlags::empty(),
+            "hi",
+            OpenFlags::CREATE,
+            DescriptorFlags::empty(),
+        )
+        .unwrap();
+    f.write(&[0; 5001], 0).unwrap();
+    unreachable!()
+}
+
+fn write_stream() {
+    preopen()
+        .open_at(
+            PathFlags::empty(),
+            "hi",
+            OpenFlags::CREATE,
+            DescriptorFlags::empty(),
+        )
+        .unwrap()
+        .write_via_stream(0)
+        .unwrap()
+        .write(&[0; 5001])
+        .unwrap();
+    unreachable!()
+}
+
+fn write_stream_blocking() {
+    preopen()
+        .open_at(
+            PathFlags::empty(),
+            "hi",
+            OpenFlags::CREATE,
+            DescriptorFlags::empty(),
+        )
+        .unwrap()
+        .write_via_stream(0)
+        .unwrap()
+        .blocking_write_and_flush(&[0; 5001])
+        .unwrap();
+    unreachable!()
+}
+
+fn mkdir() {
+    let mut name = String::new();
+    for _ in 0..5001 {
+        name.push_str("a");
+    }
+    preopen().create_directory_at(&name).unwrap();
+    unreachable!()
+}
+
+fn resolve() {
+    let network = wasip2::sockets::instance_network::instance_network();
+    let mut name = String::new();
+    for _ in 0..5001 {
+        name.push_str("a");
+    }
+    wasip2::sockets::ip_name_lookup::resolve_addresses(&network, &name).unwrap();
+    unreachable!();
+}
+
+fn udp_socket() -> wasip2::sockets::udp::UdpSocket {
+    let socket =
+        wasip2::sockets::udp_create_socket::create_udp_socket(IpAddressFamily::Ipv4).unwrap();
+    let network = wasip2::sockets::instance_network::instance_network();
+
+    socket
+        .start_bind(
+            &network,
+            IpSocketAddress::Ipv4(Ipv4SocketAddress {
+                address: (127, 0, 0, 1),
+                port: 0,
+            }),
+        )
+        .unwrap();
+    socket.finish_bind().unwrap();
+    socket
+}
+
+fn udp_send_many() {
+    let socket = udp_socket();
+    let (_incoming, outgoing) = socket.stream(None).unwrap();
+    let mut dgrams = Vec::new();
+
+    for _ in 0..5000 {
+        dgrams.push(wasip2::sockets::udp::OutgoingDatagram {
+            data: Vec::new(),
+            remote_address: None,
+        });
+    }
+
+    outgoing.send(&dgrams).unwrap();
+    unreachable!()
+}
+
+fn udp_send_big() {
+    let socket = udp_socket();
+    let (_incoming, outgoing) = socket.stream(None).unwrap();
+    let mut dgrams = Vec::new();
+
+    for _ in 0..2 {
+        dgrams.push(wasip2::sockets::udp::OutgoingDatagram {
+            data: vec![0; 2500],
+            remote_address: None,
+        });
+    }
+
+    outgoing.send(&dgrams).unwrap();
+    unreachable!()
+}
+
+fn write_zeroes() {
+    preopen()
+        .open_at(
+            PathFlags::empty(),
+            "hi",
+            OpenFlags::CREATE,
+            DescriptorFlags::empty(),
+        )
+        .unwrap()
+        .write_via_stream(0)
+        .unwrap()
+        .write_zeroes(u64::MAX)
+        .unwrap();
+    unreachable!()
+}
+
+fn write_stream_buffer_too_large() {
+    preopen()
+        .open_at(
+            PathFlags::empty(),
+            "hi",
+            OpenFlags::CREATE,
+            DescriptorFlags::empty(),
+        )
+        .unwrap()
+        .write_via_stream(0)
+        .unwrap()
+        .blocking_write_and_flush(&[0; 5000])
+        .unwrap();
+    unreachable!()
+}
+
+fn write_zeroes_buffer_too_large() {
+    preopen()
+        .open_at(
+            PathFlags::empty(),
+            "hi",
+            OpenFlags::CREATE,
+            DescriptorFlags::empty(),
+        )
+        .unwrap()
+        .write_via_stream(0)
+        .unwrap()
+        .blocking_write_zeroes_and_flush(5000)
+        .unwrap();
+    unreachable!()
+}
+
+fn read_file_big() {
+    preopen()
+        .open_at(
+            PathFlags::empty(),
+            "1mb",
+            OpenFlags::empty(),
+            DescriptorFlags::empty(),
+        )
+        .unwrap()
+        .read_via_stream(0)
+        .unwrap()
+        .blocking_read(u64::MAX)
+        .unwrap();
+}
+
+fn read_tcp_big() {
+    let server =
+        wasip2::sockets::tcp_create_socket::create_tcp_socket(IpAddressFamily::Ipv4).unwrap();
+    let client =
+        wasip2::sockets::tcp_create_socket::create_tcp_socket(IpAddressFamily::Ipv4).unwrap();
+
+    server
+        .start_bind(
+            &wasip2::sockets::instance_network::instance_network(),
+            IpSocketAddress::Ipv4(Ipv4SocketAddress {
+                address: (127, 0, 0, 1),
+                port: 0,
+            }),
+        )
+        .unwrap();
+    server.finish_bind().unwrap();
+    server.start_listen().unwrap();
+    server.finish_listen().unwrap();
+
+    client
+        .start_connect(
+            &wasip2::sockets::instance_network::instance_network(),
+            server.local_address().unwrap(),
+        )
+        .unwrap();
+    client.subscribe().block();
+    let (input, _output) = client.finish_connect().unwrap();
+
+    {
+        server.subscribe().block();
+        let (socket, input, output) = server.accept().unwrap();
+        drop((input, output));
+        drop(socket);
+    }
+
+    match input.blocking_read(u64::MAX) {
+        Err(wasip2::io::streams::StreamError::Closed) => {}
+        other => panic!("unexpected result: {other:?}"),
+    }
+}

@@ -0,0 +1,33 @@
+fn main() {
+    let fields = wasip2::http::types::Fields::new();
+
+    match std::env::args().nth(1).as_deref() {
+        Some("append") => {
+            for i in 0.. {
+                if fields.append(&format!("a{i}"), b"a").is_err() {
+                    break;
+                }
+            }
+        }
+        Some("append-empty") => {
+            for i in 0.. {
+                if fields.append(&format!("a{i}"), b"").is_err() {
+                    break;
+                }
+            }
+        }
+        Some("append-same") => loop {
+            if fields.append("a", b"b").is_err() {
+                break;
+            }
+        },
+        Some("append-same-empty") => loop {
+            if fields.append("a", b"").is_err() {
+                break;
+            }
+        },
+        other => panic!("unknown test {other:?}"),
+    }
+
+    unreachable!();
+}

@@ -0,0 +1,7 @@
+use test_programs::wasi::clocks::monotonic_clock::subscribe_duration;
+
+fn main() {
+    loop {
+        std::mem::forget(subscribe_duration(1_000_000));
+    }
+}

@@ -0,0 +1,6 @@
+fn main() {
+    let mut buf = Vec::new();
+    for _ in 0..100 {
+        buf.push(wasip2::clocks::monotonic_clock::subscribe_duration(0));
+    }
+}

@@ -1,16 +1,23 @@
+use std::mem::MaybeUninit;
 use test_programs::wasi::random;
 
 fn main() {
-    let mut bytes = [0_u8; 256];
+    let p1_random_size: usize = std::env::var("TEST_P1_RANDOM_LEN")
+        .map(|v| v.parse().expect("TEST_P1_RANDOM_LEN should be a usize"))
+        .unwrap_or(256);
+    let mut bytes: Box<[MaybeUninit<u8>]> = Box::new_uninit_slice(p1_random_size);
     unsafe {
-        wasip1::random_get(bytes.as_mut_ptr(), bytes.len()).unwrap();
+        wasip1::random_get(bytes.as_mut_ptr() as *mut u8, bytes.len()).unwrap();
     }
 
-    assert!(bytes.iter().any(|x| *x != 0));
+    assert!(bytes.iter().any(|x| unsafe { x.assume_init() } != 0));
 
+    let p2_random_size: u64 = std::env::var("TEST_P2_RANDOM_LEN")
+        .map(|v| v.parse().expect("TEST_P2_RANDOM_LEN should be a u64"))
+        .unwrap_or(256);
     // Acquired random bytes should be of the expected length.
-    let array = random::random::get_random_bytes(100);
-    assert_eq!(array.len(), 100);
+    let array = random::random::get_random_bytes(p2_random_size);
+    assert_eq!(array.len(), p2_random_size as usize);
 
     // It shouldn't take 100+ tries to get a nonzero random integer.
     for i in 0.. {
@@ -27,9 +34,15 @@ fn main() {
     assert_eq!(a1, a2);
     assert_eq!(b1, b2);
 
+    let p2_insecure_random_size: u64 = std::env::var("TEST_P2_INSECURE_RANDOM_LEN")
+        .map(|v| {
+            v.parse()
+                .expect("TEST_P2_INSECURE_RANDOM_LEN should be a u64")
+        })
+        .unwrap_or(256);
     // Acquired random bytes should be of the expected length.
-    let array = random::insecure::get_insecure_random_bytes(100);
-    assert_eq!(array.len(), 100);
+    let array = random::insecure::get_insecure_random_bytes(p2_insecure_random_size);
+    assert_eq!(array.len(), p2_insecure_random_size as usize);
 
     // It shouldn't take 100+ tries to get a nonzero random integer.
     for i in 0.. {

@@ -27,7 +27,7 @@ log = { workspace = true }
 async-trait = { workspace = true }
 
 # Optional, enabled by wasmtime feature:
-wasmtime = { workspace = true, optional = true, features = ['runtime'] }
+wasmtime = { workspace = true, optional = true, features = ['runtime', 'component-model'] }
 # Optional, enabled by sync feature:
 cap-fs-ext = { workspace = true, optional = true }
 cap-time-ext = { workspace = true, optional = true }

@@ -5,8 +5,8 @@ use crate::{
         subscription::{RwEventFlags, Subscription},
     },
 };
-use std::future::Future;
-use std::pin::Pin;
+use std::future::{self, Future};
+use std::pin::{Pin, pin};
 use std::task::{Context, Poll as FPoll};
 
 struct FirstReady<'a, T>(Vec<Pin<Box<dyn Future<Output = T> + Send + 'a>>>);
@@ -83,13 +83,20 @@ pub async fn poll_oneoff<'a>(poll: &mut Poll<'a>) -> Result<(), Error> {
             Subscription::MonotonicClock { .. } => unreachable!(),
         }
     }
-    if let Some(Some(remaining_duration)) = duration {
-        match tokio::time::timeout(remaining_duration, futures).await {
+    match duration {
+        Some(Some(remaining)) => match tokio::time::timeout(remaining, futures).await {
             Ok(r) => r?,
             Err(_deadline_elapsed) => {}
+        },
+        Some(None) => {
+            let mut futures = pin!(futures);
+            future::poll_fn(|cx| match futures.as_mut().poll(cx) {
+                FPoll::Ready(e) => FPoll::Ready(e),
+                FPoll::Pending => FPoll::Ready(Ok(())),
+            })
+            .await?
         }
-    } else {
-        futures.await?;
+        None => futures.await?,
     }
 
     Ok(())

@@ -289,9 +289,7 @@ async fn p1_file_write() {
 async fn p1_path_open_lots() {
     run(P1_PATH_OPEN_LOTS, true).await.unwrap()
 }
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p1_cli_much_stdout() {}
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
+async fn p1_sleep_quickly_but_lots() {
+    run(P1_SLEEP_QUICKLY_BUT_LOTS, true).await.unwrap()
+}

@@ -281,9 +281,7 @@ fn p1_file_write() {
 fn p1_path_open_lots() {
     run(P1_PATH_OPEN_LOTS, true).unwrap()
 }
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p1_cli_much_stdout() {}
+#[test_log::test]
+fn p1_sleep_quickly_but_lots() {
+    run(P1_SLEEP_QUICKLY_BUT_LOTS, true).unwrap()
+}

@@ -24,6 +24,7 @@ pub type HyperOutgoingBody = UnsyncBoxBody<Bytes, types::ErrorCode>;
 #[derive(Debug)]
 pub struct HostIncomingBody {
     body: IncomingBodyState,
+    field_size_limit: usize,
     /// An optional worker task to keep alive while this body is being read.
     /// This ensures that if the parent of this body is dropped before the body
     /// then the backing data behind this worker is kept alive.
@@ -32,10 +33,15 @@ pub struct HostIncomingBody {
 
 impl HostIncomingBody {
     /// Create a new `HostIncomingBody` with the given `body` and a per-frame timeout
-    pub fn new(body: HyperIncomingBody, between_bytes_timeout: Duration) -> HostIncomingBody {
+    pub fn new(
+        body: HyperIncomingBody,
+        between_bytes_timeout: Duration,
+        field_size_limit: usize,
+    ) -> HostIncomingBody {
         let body = BodyWithTimeout::new(body, between_bytes_timeout);
         HostIncomingBody {
             body: IncomingBodyState::Start(body),
+            field_size_limit,
             worker: None,
         }
     }
@@ -155,7 +161,7 @@ enum StreamEnd {
 
     /// Body was completely read and trailers were read. Here are the trailers.
     /// Note that `None` means that the body finished without trailers.
-    Trailers(Option<FieldMap>),
+    Trailers(Option<http::HeaderMap>),
 }
 
 /// The concrete type behind the `wasi:io/streams.input-stream` resource returned
@@ -339,14 +345,15 @@ impl Pollable for HostFutureTrailers {
             match rx.await {
                 // Trailers were read for us and here they are, so store the
                 // result.
-                Ok(StreamEnd::Trailers(t)) => *self = Self::Done(Ok(t)),
-
+                Ok(StreamEnd::Trailers(Some(t))) => {
+                    *self = Self::Done(Ok(Some(FieldMap::new(t, body.field_size_limit))));
+                }
                 // The body wasn't fully read and was dropped before trailers
                 // were reached. It's up to us now to complete the body.
                 Ok(StreamEnd::Remaining(b)) => body.body = IncomingBodyState::Start(b),
 
                 // This means there were no trailers present.
-                Err(_) => {
+                Ok(StreamEnd::Trailers(None)) | Err(_) => {
                     *self = HostFutureTrailers::Done(Ok(None));
                 }
             }
@@ -370,8 +377,8 @@ impl Pollable for HostFutureTrailers {
                 Some(Ok(frame)) => {
                     // If this frame is a data frame ignore it as we're only
                     // interested in trailers.
-                    if let Ok(headers) = frame.into_trailers() {
-                        break Ok(Some(headers));
+                    if let Ok(header_map) = frame.into_trailers() {
+                        break Ok(Some(FieldMap::new(header_map, body.field_size_limit)));
                     }
                 }
             }
@@ -521,7 +528,7 @@ impl HostOutgoingBody {
         }
 
         let message = if let Some(ts) = trailers {
-            FinishMessage::Trailers(ts)
+            FinishMessage::Trailers(ts.into_inner())
         } else {
             FinishMessage::Finished
         };

@@ -79,7 +79,7 @@ where
 
         builder = builder.uri(uri.build().map_err(http_request_error)?);
 
-        for (k, v) in req.headers.iter() {
+        for (k, v) in req.headers.as_ref().iter() {
             builder = builder.header(k, v);
         }
 

@@ -3,11 +3,12 @@
 use crate::bindings::http::types::{self, Headers, Method, Scheme, StatusCode, Trailers};
 use crate::body::{HostFutureTrailers, HostIncomingBody, HostOutgoingBody, StreamContext};
 use crate::types::{
-    FieldMap, HostFields, HostFutureIncomingResponse, HostIncomingRequest, HostIncomingResponse,
-    HostOutgoingRequest, HostOutgoingResponse, HostResponseOutparam, remove_forbidden_headers,
+    FieldMap, FieldSizeLimitError, HostFields, HostFutureIncomingResponse, HostIncomingRequest,
+    HostIncomingResponse, HostOutgoingRequest, HostOutgoingResponse, HostResponseOutparam,
+    remove_forbidden_headers,
 };
 use crate::{HttpError, HttpResult, WasiHttpImpl, WasiHttpView, get_content_length};
-use anyhow::{Context, anyhow};
+use anyhow::{Context, anyhow, bail};
 use std::any::Any;
 use std::str::FromStr;
 use wasmtime::component::{Resource, ResourceTable, ResourceTableError};
@@ -80,10 +81,11 @@ where
     T: WasiHttpView,
 {
     fn new(&mut self) -> wasmtime::Result<Resource<HostFields>> {
+        let limit = self.ctx().field_size_limit;
         let id = self
             .table()
             .push(HostFields::Owned {
-                fields: hyper::HeaderMap::new(),
+                fields: FieldMap::empty(limit),
             })
             .context("[new_fields] pushing fields")?;
 
@@ -114,6 +116,14 @@ where
             fields.append(header, value);
         }
 
+        let size = FieldMap::content_size(&fields);
+        if size > self.ctx().field_size_limit {
+            bail!(FieldSizeLimitError {
+                size,
+                limit: self.ctx().field_size_limit,
+            });
+        }
+        let fields = FieldMap::new(fields, self.ctx().field_size_limit);
         let id = self
             .table()
             .push(HostFields::Owned { fields })
@@ -141,11 +151,12 @@ where
             Err(_) => return Ok(vec![]),
         };
 
-        if !fields.contains_key(&header) {
+        if !fields.as_ref().contains_key(&header) {
             return Ok(vec![]);
         }
 
         let res = fields
+            .as_ref()
             .get_all(&header)
             .into_iter()
             .map(|val| val.as_bytes().to_owned())
@@ -157,7 +168,7 @@ where
         let fields = get_fields(self.table(), &fields).context("[fields_get] getting fields")?;
 
         match hyper::header::HeaderName::from_bytes(name.as_bytes()) {
-            Ok(header) => Ok(fields.contains_key(&header)),
+            Ok(header) => Ok(fields.as_ref().contains_key(&header)),
             Err(_) => Ok(false),
         }
     }
@@ -185,14 +196,18 @@ where
             }
         }
 
-        Ok(get_fields_mut(self.table(), &fields)
+        match get_fields_mut(self.table(), &fields)
             .context("[fields_set] getting mutable fields")?
-            .map(|fields| {
-                fields.remove(&header);
+        {
+            Ok(fields) => {
+                fields.remove_all(&header);
                 for value in values {
-                    fields.append(&header, value);
+                    fields.append(&header, value)?;
                 }
-            }))
+                Ok(Ok(()))
+            }
+            Err(e) => Ok(Err(e)),
+        }
     }
 
     fn delete(
@@ -210,7 +225,7 @@ where
         }
 
         Ok(get_fields_mut(self.table(), &fields)?.map(|fields| {
-            fields.remove(header);
+            fields.remove_all(&header);
         }))
     }
 
@@ -234,18 +249,23 @@ where
             Err(_) => return Ok(Err(types::HeaderError::InvalidSyntax)),
         };
 
-        Ok(get_fields_mut(self.table(), &fields)
+        match get_fields_mut(self.table(), &fields)
             .context("[fields_append] getting mutable fields")?
-            .map(|fields| {
-                fields.append(header, value);
-            }))
+        {
+            Ok(fields) => {
+                fields.append(&header, value)?;
+                Ok(Ok(()))
+            }
+            Err(e) => Ok(Err(e)),
+        }
     }
 
     fn entries(
         &mut self,
         fields: Resource<HostFields>,
     ) -> wasmtime::Result<Vec<(String, Vec<u8>)>> {
         Ok(get_fields(self.table(), &fields)?
+            .as_ref()
             .iter()
             .map(|(name, value)| (name.as_str().to_owned(), value.as_bytes().to_owned()))
             .collect())
@@ -270,7 +290,7 @@ where
     T: WasiHttpView,
 {
     fn method(&mut self, id: Resource<HostIncomingRequest>) -> wasmtime::Result<Method> {
-        let method = self.table().get(&id)?.parts.method.clone();
+        let method = self.table().get(&id)?.method.clone();
         Ok(method.into())
     }
     fn path_with_query(
@@ -279,7 +299,6 @@ where
     ) -> wasmtime::Result<Option<String>> {
         let req = self.table().get(&id)?;
         Ok(req
-            .parts
             .uri
             .path_and_query()
             .map(|path_and_query| path_and_query.as_str().to_owned()))
@@ -300,11 +319,7 @@ where
         let _ = self.table().get(&id)?;
 
         fn get_fields(elem: &mut dyn Any) -> &mut FieldMap {
-            &mut elem
-                .downcast_mut::<HostIncomingRequest>()
-                .unwrap()
-                .parts
-                .headers
+            &mut elem.downcast_mut::<HostIncomingRequest>().unwrap().headers
         }
 
         let headers = self.table().push_child(
@@ -376,7 +391,7 @@ where
             return Ok(Err(()));
         }
 
-        let size = match get_content_length(&req.headers) {
+        let size = match get_content_length(req.headers.as_ref()) {
             Ok(size) => size,
             Err(..) => return Ok(Err(())),
         };
@@ -653,7 +668,7 @@ where
     {
         let trailers = self.table().get_mut(&id)?;
         match trailers {
-            HostFutureTrailers::Waiting(_) => return Ok(None),
+            HostFutureTrailers::Waiting { .. } => return Ok(None),
             HostFutureTrailers::Consumed => return Ok(Some(Err(()))),
             HostFutureTrailers::Done(_) => {}
         };
@@ -742,7 +757,7 @@ where
             return Ok(Err(()));
         }
 
-        let size = match get_content_length(&resp.headers) {
+        let size = match get_content_length(resp.headers.as_ref()) {
             Ok(size) => size,
             Err(..) => return Ok(Err(())),
         };
@@ -821,6 +836,7 @@ where
     ) -> wasmtime::Result<
         Option<Result<Result<Resource<HostIncomingResponse>, types::ErrorCode>, ()>>,
     > {
+        let field_size_limit = self.ctx().field_size_limit;
         let resp = self.table().get_mut(&id)?;
 
         match resp {
@@ -841,15 +857,17 @@ where
                 Ok(Err(e)) => return Ok(Some(Ok(Err(e)))),
             };
 
-        let (mut parts, body) = resp.resp.into_parts();
+        let (parts, body) = resp.resp.into_parts();
 
-        remove_forbidden_headers(self, &mut parts.headers);
+        let mut headers = FieldMap::new(parts.headers, field_size_limit);
+        remove_forbidden_headers(self, &mut headers);
 
         let resp = self.table().push(HostIncomingResponse {
             status: parts.status.as_u16(),
-            headers: parts.headers,
+            headers,
             body: Some({
-                let mut body = HostIncomingBody::new(body, resp.between_bytes_timeout);
+                let mut body =
+                    HostIncomingBody::new(body, resp.between_bytes_timeout, field_size_limit);
                 if let Some(worker) = resp.worker {
                     body.retain_worker(worker);
                 }

@@ -7,11 +7,13 @@ use crate::{
 };
 use anyhow::bail;
 use bytes::Bytes;
+use http::header::{HeaderMap, HeaderName, HeaderValue};
 use http_body_util::BodyExt;
 use hyper::body::Body;
-use hyper::header::HeaderName;
 use std::any::Any;
+use std::fmt;
 use std::time::Duration;
+use wasmtime::Result;
 use wasmtime::component::{Resource, ResourceTable};
 use wasmtime_wasi::p2::Pollable;
 use wasmtime_wasi::runtime::AbortOnDropJoinHandle;
@@ -24,16 +26,39 @@ use {
     tokio::time::timeout,
 };
 
+/// Default maximum size for the contents of a fields resource.
+///
+/// Typically, HTTP proxies limit headers to 8k. This number is higher than that
+/// because it not only includes the wire-size of headers but it additionally
+/// includes factors for the in-memory representation of `HeaderMap`. This is in
+/// theory high enough that no one runs into it but low enough such that a
+/// completely full `HeaderMap` doesn't break the bank in terms of memory
+/// consumption.
+const DEFAULT_FIELD_SIZE_LIMIT: usize = 2 << 30;
+
 /// Capture the state necessary for use in the wasi-http API implementation.
 #[derive(Debug)]
 pub struct WasiHttpCtx {
-    _priv: (),
+    pub(crate) field_size_limit: usize,
 }
 
 impl WasiHttpCtx {
     /// Create a new context.
     pub fn new() -> Self {
-        Self { _priv: () }
+        Self {
+            field_size_limit: DEFAULT_FIELD_SIZE_LIMIT,
+        }
+    }
+
+    /// Set the maximum size for any fields resources created by this context.
+    ///
+    /// The limit specified here is roughly a byte limit for the size of the
+    /// in-memory representation of headers. This means that the limit needs to
+    /// be larger than the literal representation of headers on the wire to
+    /// account for in-memory Rust-side data structures representing the header
+    /// names/values/etc.
+    pub fn set_field_size_limit(&mut self, limit: usize) {
+        self.field_size_limit = limit;
     }
 }
 
@@ -95,14 +120,17 @@ pub trait WasiHttpView {
         B: Body<Data = Bytes, Error = hyper::Error> + Send + 'static,
         Self: Sized,
     {
+        let field_size_limit = self.ctx().field_size_limit;
         let (parts, body) = req.into_parts();
         let body = body.map_err(crate::hyper_response_error).boxed_unsync();
         let body = HostIncomingBody::new(
             body,
             // TODO: this needs to be plumbed through
             std::time::Duration::from_millis(600 * 1000),
+            field_size_limit,
         );
-        let incoming_req = HostIncomingRequest::new(self, parts, scheme, Some(body))?;
+        let incoming_req =
+            HostIncomingRequest::new(self, parts, scheme, Some(body), field_size_limit)?;
         Ok(self.table().push(incoming_req)?)
     }
 
@@ -305,12 +333,9 @@ pub const DEFAULT_FORBIDDEN_HEADERS: [http::header::HeaderName; 9] = [
     HeaderName::from_static("http2-settings"),
 ];
 
-/// Removes forbidden headers from a [`hyper::HeaderMap`].
-pub(crate) fn remove_forbidden_headers(
-    view: &mut dyn WasiHttpView,
-    headers: &mut hyper::HeaderMap,
-) {
-    let forbidden_keys = Vec::from_iter(headers.keys().filter_map(|name| {
+/// Removes forbidden headers from a [`FieldMap`].
+pub(crate) fn remove_forbidden_headers(view: &mut dyn WasiHttpView, headers: &mut FieldMap) {
+    let forbidden_keys = Vec::from_iter(headers.as_ref().keys().filter_map(|name| {
         if view.is_forbidden_header(name) {
             Some(name.clone())
         } else {
@@ -319,7 +344,7 @@ pub(crate) fn remove_forbidden_headers(
     }));
 
     for name in forbidden_keys {
-        headers.remove(name);
+        headers.remove_all(&name);
     }
 }
 
@@ -533,7 +558,9 @@ impl TryInto<http::Method> for types::Method {
 /// The concrete type behind a `wasi:http/types.incoming-request` resource.
 #[derive(Debug)]
 pub struct HostIncomingRequest {
-    pub(crate) parts: http::request::Parts,
+    pub(crate) method: http::method::Method,
+    pub(crate) uri: http::uri::Uri,
+    pub(crate) headers: FieldMap,
     pub(crate) scheme: Scheme,
     pub(crate) authority: String,
     /// The body of the incoming request.
@@ -544,9 +571,10 @@ impl HostIncomingRequest {
     /// Create a new `HostIncomingRequest`.
     pub fn new(
         view: &mut dyn WasiHttpView,
-        mut parts: http::request::Parts,
+        parts: http::request::Parts,
         scheme: Scheme,
         body: Option<HostIncomingBody>,
+        field_size_limit: usize,
     ) -> anyhow::Result<Self> {
         let authority = match parts.uri.authority() {
             Some(authority) => authority.to_string(),
@@ -556,9 +584,13 @@ impl HostIncomingRequest {
             },
         };
 
-        remove_forbidden_headers(view, &mut parts.headers);
+        let mut headers = FieldMap::new(parts.headers, field_size_limit);
+        remove_forbidden_headers(view, &mut headers);
+
         Ok(Self {
-            parts,
+            method: parts.method,
+            uri: parts.uri,
+            headers,
             authority,
             scheme,
             body,
@@ -593,7 +625,7 @@ impl TryFrom<HostOutgoingResponse> for hyper::Response<HyperOutgoingBody> {
 
         let mut builder = hyper::Response::builder().status(resp.status);
 
-        *builder.headers_mut().unwrap() = resp.headers;
+        *builder.headers_mut().unwrap() = resp.headers.map;
 
         match resp.body {
             Some(body) => builder.body(body),
@@ -667,8 +699,130 @@ pub enum HostFields {
     },
 }
 
-/// An owned version of `HostFields`
-pub type FieldMap = hyper::HeaderMap;
+/// An owned version of `HostFields`. A wrapper on http `HeaderMap` that
+/// keeps a running tally of memory consumed by header names and values.
+#[derive(Debug, Clone)]
+pub struct FieldMap {
+    map: HeaderMap,
+    limit: usize,
+    size: usize,
+}
+
+/// Error given when a `FieldMap` has exceeded the size limit.
+#[derive(Debug)]
+pub struct FieldSizeLimitError {
+    /// The erroring `FieldMap` operation would require this content size
+    pub(crate) size: usize,
+    /// The limit set on `FieldMap` content size
+    pub(crate) limit: usize,
+}
+impl fmt::Display for FieldSizeLimitError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "Field size limit {} exceeded: {}", self.limit, self.size)
+    }
+}
+impl std::error::Error for FieldSizeLimitError {}
+
+impl FieldMap {
+    /// Construct a `FieldMap` from a `HeaderMap` and a size limit.
+    ///
+    /// Construction with a `HeaderMap` which exceeds the size limit is
+    /// allowed, but subsequent operations to expand the resource use will
+    /// fail.
+    pub fn new(map: HeaderMap, limit: usize) -> Self {
+        let size = Self::content_size(&map);
+        Self { map, size, limit }
+    }
+    /// Construct an empty `FieldMap`
+    pub fn empty(limit: usize) -> Self {
+        Self {
+            map: HeaderMap::new(),
+            size: 0,
+            limit,
+        }
+    }
+    /// Get the `HeaderMap` out of the `FieldMap`
+    pub fn into_inner(self) -> HeaderMap {
+        self.map
+    }
+    /// Calculate the content size of a `HeaderMap`. This is a sum of the size
+    /// of all of the keys and all of the values.
+    pub(crate) fn content_size(map: &HeaderMap) -> usize {
+        let mut sum = 0;
+        for key in map.keys() {
+            sum += header_name_size(key);
+        }
+        for value in map.values() {
+            sum += header_value_size(value);
+        }
+        sum
+    }
+    /// Remove all values associated with a key in a map.
+    ///
+    /// Returns an empty list if the key is not already present within the map.
+    pub fn remove_all(&mut self, key: &HeaderName) -> Vec<HeaderValue> {
+        use http::header::Entry;
+        match self.map.try_entry(key) {
+            Ok(Entry::Vacant { .. }) | Err(_) => Vec::new(),
+            Ok(Entry::Occupied(e)) => {
+                let (name, value_drain) = e.remove_entry_mult();
+                let mut removed = header_name_size(&name);
+                let values = value_drain.collect::<Vec<_>>();
+                for v in values.iter() {
+                    removed += header_value_size(v);
+                }
+                self.size -= removed;
+                values
+            }
+        }
+    }
+    /// Add a value associated with a key to the map.
+    ///
+    /// If `key` is already present within the map then `value` is appended to
+    /// the list of values it already has.
+    pub fn append(&mut self, key: &HeaderName, value: HeaderValue) -> Result<bool> {
+        let key_size = header_name_size(key);
+        let val_size = header_value_size(&value);
+        let new_size = if !self.map.contains_key(key) {
+            self.size + key_size + val_size
+        } else {
+            self.size + val_size
+        };
+        if new_size > self.limit {
+            bail!(FieldSizeLimitError {
+                limit: self.limit,
+                size: new_size
+            })
+        }
+        self.size = new_size;
+        Ok(self.map.try_append(key, value)?)
+    }
+}
+
+/// Returns the size, in accounting cost, to consider for `name`.
+///
+/// This includes both the byte length of the `name` itself as well as the size
+/// of the data structure itself as it'll reside within a `HeaderMap`.
+fn header_name_size(name: &HeaderName) -> usize {
+    name.as_str().len() + size_of::<HeaderName>()
+}
+
+/// Same as `header_name_size`, but for values.
+///
+/// This notably includes the size of `HeaderValue` itself to ensure that all
+/// headers have a nonzero size as otherwise this would never limit addition of
+/// an empty header value.
+fn header_value_size(value: &HeaderValue) -> usize {
+    value.len() + size_of::<HeaderValue>()
+}
+
+// We impl AsRef, but not AsMut, because any modifications of the
+// underlying HeaderMap must account for changes in size
+impl AsRef<HeaderMap> for FieldMap {
+    fn as_ref(&self) -> &HeaderMap {
+        &self.map
+    }
+}
 
 /// A handle to a future incoming response.
 pub type FutureIncomingResponseHandle =

@@ -123,6 +123,7 @@ async fn run_wasi_http(
     send_request: Option<RequestSender>,
     rejected_authority: Option<String>,
     early_drop: bool,
+    field_size_limit: Option<usize>,
 ) -> anyhow::Result<Result<hyper::Response<Collected<Bytes>>, ErrorCode>> {
     let stdout = MemoryOutputPipe::new(4096);
     let stderr = MemoryOutputPipe::new(4096);
@@ -140,7 +141,10 @@ async fn run_wasi_http(
     builder.stdout(stdout.clone());
     builder.stderr(stderr.clone());
     let wasi = builder.build();
-    let http = WasiHttpCtx::new();
+    let mut http = WasiHttpCtx::new();
+    if let Some(limit) = field_size_limit {
+        http.set_field_size_limit(limit);
+    }
     let ctx = Ctx {
         table,
         wasi,
@@ -153,15 +157,22 @@ async fn run_wasi_http(
     let mut store = Store::new(&engine, ctx);
 
     let mut linker = Linker::new(&engine);
-    wasmtime_wasi_http::add_to_linker_async(&mut linker)?;
+    wasmtime_wasi_http::add_to_linker_async(&mut linker).context("add crate to linker")?;
     let proxy =
         wasmtime_wasi_http::bindings::Proxy::instantiate_async(&mut store, &component, &linker)
-            .await?;
+            .await
+            .context("instantiate proxy")?;
 
-    let req = store.data_mut().new_incoming_request(Scheme::Http, req)?;
+    let req = store
+        .data_mut()
+        .new_incoming_request(Scheme::Http, req)
+        .context("new incoming request")?;
 
     let (sender, receiver) = tokio::sync::oneshot::channel();
-    let out = store.data_mut().new_response_outparam(sender)?;
+    let out = store
+        .data_mut()
+        .new_response_outparam(sender)
+        .context("new response outparam")?;
 
     let receiver = if early_drop {
         // Drop the receiver early, emulating a host event like
@@ -176,7 +187,8 @@ async fn run_wasi_http(
         proxy
             .wasi_http_incoming_handler()
             .call_handle(&mut store, req, out)
-            .await?;
+            .await
+            .context("calling incoming handler")?;
 
         Ok::<_, anyhow::Error>(())
     });
@@ -185,7 +197,7 @@ async fn run_wasi_http(
         let resp = match r.await {
             Ok(Ok(resp)) => {
                 let (parts, body) = resp.into_parts();
-                let collected = BodyExt::collect(body).await?;
+                let collected = BodyExt::collect(body).await.context("collecting body")?;
                 Some(Ok(hyper::Response::from_parts(parts, collected)))
             }
             Ok(Err(e)) => Some(Err(e)),
@@ -197,9 +209,9 @@ async fn run_wasi_http(
 
         // Now that the response has been processed, we can wait on the wasm to
         // finish without deadlocking.
-        handle.await.context("Component execution")?;
+        handle.await.context("awaiting execution")?;
 
-        Ok(resp.expect("wasm never called set-response-outparam"))
+        Ok(resp.context("wasm never called set-response-outparam")?)
     } else {
         handle.await.context("Component execution")?;
         Ok(Err(ErrorCode::HttpResponseTimeout))
@@ -219,6 +231,7 @@ async fn wasi_http_proxy_tests() -> anyhow::Result<()> {
         None,
         None,
         false,
+        None,
     )
     .await?;
 
@@ -351,6 +364,7 @@ async fn do_wasi_http_hash_all(override_send_request: bool) -> Result<()> {
         send_request,
         None,
         false,
+        None,
     )
     .await??;
 
@@ -399,6 +413,7 @@ async fn wasi_http_hash_all_with_reject() -> Result<()> {
         None,
         Some("forbidden.com".to_string()),
         false,
+        None,
     )
     .await??;
 
@@ -519,6 +534,7 @@ async fn do_wasi_http_echo(uri: &str, url_header: Option<&str>) -> Result<()> {
         None,
         None,
         false,
+        None,
     )
     .await??;
 
@@ -551,6 +567,7 @@ async fn wasi_http_without_port() -> Result<()> {
         None,
         None,
         false,
+        None,
     )
     .await??;
 
@@ -574,6 +591,7 @@ async fn wasi_http_no_trap_on_early_drop() -> Result<()> {
         None,
         None,
         true,
+        None,
     )
     .await?;
 
@@ -583,3 +601,98 @@ async fn wasi_http_no_trap_on_early_drop() -> Result<()> {
         panic!("test expects an error");
     }
 }
+
+#[test_log::test(tokio::test)]
+async fn wasi_http_fields_limit_incoming_request() -> Result<()> {
+    use crate::p2::types::FieldSizeLimitError;
+    use http::{HeaderName, HeaderValue, Request};
+    use http_body_util::combinators::BoxBody;
+    use hyper::Error;
+
+    fn request_with_header_size(uri: &str, total: usize) -> Request<BoxBody<Bytes, Error>> {
+        let mut builder = hyper::Request::builder().uri(uri).method(http::Method::GET);
+
+        let headers = builder.headers_mut().expect("builder error");
+        let chunks = total / 10;
+        let remainder = total % 10;
+        for chunk in 0..chunks {
+            let mut v = format!("v{chunk:04}");
+            if chunk == 0 {
+                for _ in 0..remainder {
+                    v.push('x');
+                }
+            }
+            headers.insert(
+                HeaderName::from_bytes(format!("k{chunk:04}").as_bytes())
+                    .expect("valid header name"),
+                HeaderValue::from_str(&v).expect("valid header value"),
+            );
+        }
+        builder
+            .body(body::empty())
+            .expect("complete building request")
+    }
+
+    let resp = run_wasi_http(
+        test_programs_artifacts::P2_API_PROXY_COMPONENT,
+        request_with_header_size("http://host/", 256),
+        None,
+        None,
+        false,
+        Some(255),
+    )
+    .await
+    .context("request with headers on wire over the size limit")??;
+    assert_eq!(resp.status(), 200);
+
+    let resp = run_wasi_http(
+        test_programs_artifacts::P2_API_PROXY_COMPONENT,
+        request_with_header_size("http://host/new_fields/50", 0),
+        None,
+        None,
+        false,
+        Some(500),
+    )
+    .await
+    .context("new_fields with size matching the size limit")??;
+    assert_eq!(resp.status(), 200);
+
+    let err = run_wasi_http(
+        test_programs_artifacts::P2_API_PROXY_COMPONENT,
+        request_with_header_size("http://host/new_fields/500", 0),
+        None,
+        None,
+        false,
+        Some(500),
+    )
+    .await
+    .err()
+    .expect("new_fields exceeding the size limit");
+    assert!(err.downcast_ref::<FieldSizeLimitError>().is_some());
+
+    let resp = run_wasi_http(
+        test_programs_artifacts::P2_API_PROXY_COMPONENT,
+        request_with_header_size("http://host/modify_fields/50", 10),
+        None,
+        None,
+        false,
+        Some(500),
+    )
+    .await??;
+    assert_eq!(resp.status(), 200);
+
+    let err = run_wasi_http(
+        test_programs_artifacts::P2_API_PROXY_COMPONENT,
+        request_with_header_size("http://host/modify_fields/50", 100),
+        None,
+        None,
+        false,
+        Some(500),
+    )
+    .await
+    .err()
+    .expect("run_wasi_http should give error");
+    assert!(err.downcast_ref::<FieldSizeLimitError>().is_some());
+
+    Ok(())
+}

@@ -5,6 +5,7 @@ use alloc::collections::BTreeMap;
 use alloc::string::String;
 use alloc::vec::Vec;
 use anyhow::{Result, anyhow};
+use bytes::Bytes;
 use core::future::Future;
 use core::pin::Pin;
 use core::task::{Context, Poll};
@@ -163,9 +164,10 @@ impl streams::HostOutputStream for ResourceTable {
             ));
         }
 
-        self.get_mut(&stream)?
-            .blocking_write_zeroes_and_flush(len as usize)
-            .await
+        // TODO: We could optimize this to not allocate one big zeroed buffer, and instead write
+        // repeatedly from a 'static buffer of zeros.
+        let bs = Bytes::from_iter(core::iter::repeat(0).take(len as usize));
+        self.get_mut(&stream)?.blocking_write_and_flush(bs).await
     }
 
     fn write_zeroes(&mut self, stream: Resource<DynOutputStream>, len: u64) -> StreamResult<()> {

@@ -233,44 +233,19 @@ pub trait OutputStream: Pollable {
     /// Returning an Err which downcasts to a [`StreamError`] will be
     /// reported to Wasm as the empty error result. Otherwise, errors will trap.
     fn write_zeroes(&mut self, nelem: usize) -> StreamResult<()> {
+        let n = self.check_write()?;
+        if nelem > n {
+            return Err(StreamError::trap(
+                "cannot write more zeroes than `check_write` allows",
+            ));
+        };
         // TODO: We could optimize this to not allocate one big zeroed buffer, and instead write
         // repeatedly from a 'static buffer of zeros.
         let bs = Bytes::from_iter(core::iter::repeat(0).take(nelem));
         self.write(bs)?;
         Ok(())
     }
 
-    /// Perform a write of up to 4096 zeroes, and then flush the stream.
-    /// Block until all of these operations are complete, or an error
-    /// occurs.
-    ///
-    /// This is a convenience wrapper around the use of `check-write`,
-    /// `subscribe`, `write-zeroes`, and `flush`, and is implemented with
-    /// the following pseudo-code:
-    ///
-    /// ```text
-    /// let pollable = this.subscribe();
-    /// while num_zeroes != 0 {
-    ///     // Wait for the stream to become writable
-    ///     pollable.block();
-    ///     let Ok(n) = this.check-write(); // eliding error handling
-    ///     let len = min(n, num_zeroes);
-    ///     this.write-zeroes(len);         // eliding error handling
-    ///     num_zeroes -= len;
-    /// }
-    /// this.flush();
-    /// // Wait for completion of `flush`
-    /// pollable.block();
-    /// // Check for any errors that arose during `flush`
-    /// let _ = this.check-write();         // eliding error handling
-    /// ```
-    async fn blocking_write_zeroes_and_flush(&mut self, nelem: usize) -> StreamResult<()> {
-        // TODO: We could optimize this to not allocate one big zeroed buffer, and instead write
-        // repeatedly from a 'static buffer of zeros.
-        let bs = Bytes::from_iter(core::iter::repeat(0).take(nelem));
-        self.blocking_write_and_flush(bs).await
-    }
-
     /// Simultaneously waits for this stream to be writable and then returns how
     /// much may be written or the last error that happened.
     async fn write_ready(&mut self) -> StreamResult<usize> {

@@ -120,7 +120,7 @@ impl Pollable for InputFile {
 
 impl InputStream for InputFile {
     fn read(&mut self, size: usize) -> StreamResult<Bytes> {
-        let mut buf = bytes::BytesMut::zeroed(size);
+        let mut buf = bytes::BytesMut::zeroed(size.min(crate::MAX_READ_SIZE_ALLOC));
         let bytes_read = self
             .file
             .read(&mut buf)

@@ -350,6 +350,20 @@ impl WasiCtxBuilder {
         self
     }
 
+    /// Configures the maximum len accepted by
+    /// `wasi:random/random.get-random-bytes` and
+    /// `wasi:random/insecure.get-insecure-random-bytes`. Calls with a len
+    /// larger than this limit will trap.
+    ///
+    /// This limit protects the host implementation from memory exhaustion from
+    /// untrusted guest input. A limit of `u64::MAX` is equivalent to no limit,
+    /// but note that this enables a guest to also force the host to attempt an
+    /// allocation of that size.
+    pub fn max_random_size(&mut self, max_size: u64) -> &mut Self {
+        self.random.max_size = max_size;
+        self
+    }
+
     /// Configures `wasi:clocks/wall-clock` to use the `clock` specified.
     ///
     /// By default the host's wall clock is used.

@@ -12,6 +12,24 @@
 //!
 //! For WASIp3, see [`p3`]. WASIp3 support is experimental, unstable and incomplete.
 
+/// The maximum size, in bytes, that this crate will allocate on the host on a
+/// per-read basis.
+///
+/// This is used to limit the size of `wasi:io/streams.input-stream#read`, for
+/// example, along with a variety of other read-style APIs. All of these APIs
+/// have the shape where the guest asks the host to read some data for it, and
+/// then it's returned. The data is allocated on the host, however, meaning that
+/// when the guest asks for an extremely large read that could cause a very
+/// large allocations on the host. For now this constant serves as a hard limit
+/// on the size of the allocation a host will make.
+///
+/// TODO: make this configurable per-I/O? Per-WASI? It's not great that this
+/// just a hard and unconfigurable limit. There are probably situations where
+/// performance will be improved if this limit were higher or removed. Doing so
+/// without exposing hosts to guest-controlled resource exhaustion is the
+/// important part, though.
+const MAX_READ_SIZE_ALLOC: usize = 64 * 1024;
+
 pub mod cli;
 pub mod clocks;
 mod ctx;

@@ -70,6 +70,9 @@ impl wiggle::GuestErrorType for types::Errno {
 }
 
 impl<T: Snapshot1 + Send> wasi_unstable::WasiUnstable for T {
+    fn set_hostcall_fuel(&mut self, fuel: usize) {
+        Snapshot1::set_hostcall_fuel(self, fuel)
+    }
     fn args_get(
         &mut self,
         memory: &mut GuestMemory<'_>,

@@ -145,6 +145,7 @@ pub struct WasiP1Ctx {
     table: ResourceTable,
     wasi: WasiCtx,
     adapter: WasiP1Adapter,
+    hostcall_fuel: usize,
 }
 
 impl WasiP1Ctx {
@@ -153,8 +154,81 @@ impl WasiP1Ctx {
             table: ResourceTable::new(),
             wasi,
             adapter: WasiP1Adapter::new(),
+            hostcall_fuel: 0,
         }
     }
+
+    fn consume_fuel(&mut self, fuel: usize) -> Result<()> {
+        if fuel > self.hostcall_fuel {
+            return Err(types::Errno::Nomem.into());
+        }
+        self.hostcall_fuel -= fuel;
+        Ok(())
+    }
+
+    /// Assumes the host is going to copy all of `array` in which case a
+    /// corresponding amount of fuel is consumed to ensure it's not too large.
+    fn consume_fuel_for_array<T>(&mut self, array: wiggle::GuestPtr<[T]>) -> Result<()> {
+        let byte_size = usize::try_from(array.len())?
+            .checked_mul(size_of::<T>())
+            .ok_or(types::Errno::Overflow)?;
+        self.consume_fuel(byte_size)
+    }
+
+    /// Returns the first non-empty buffer in `ciovs` or a single empty buffer
+    /// if they're all empty.
+    ///
+    /// Additionally consumes a corresponding amount of fuel appropriate to the
+    /// size of `ciovs` and the first non-empty array.
+    fn first_non_empty_ciovec(
+        &mut self,
+        memory: &GuestMemory<'_>,
+        ciovs: types::CiovecArray,
+    ) -> Result<GuestPtr<[u8]>> {
+        self.consume_fuel_for_array(ciovs)?;
+        for iov in ciovs.iter() {
+            let iov = memory.read(iov?)?;
+            if iov.buf_len == 0 {
+                continue;
+            }
+            let ret = iov.buf.as_array(iov.buf_len);
+            self.consume_fuel_for_array(ret)?;
+            return Ok(ret);
+        }
+        Ok(GuestPtr::new((0, 0)))
+    }
+
+    /// Returns the first non-empty buffer in `iovs` or a single empty buffer if
+    /// they're all empty.
+    ///
+    /// Additionally consumes a corresponding amount of fuel appropriate to the
+    /// size of `ciovs` and the first non-empty array.
+    fn first_non_empty_iovec(
+        &mut self,
+        memory: &GuestMemory<'_>,
+        iovs: types::IovecArray,
+    ) -> Result<GuestPtr<[u8]>> {
+        self.consume_fuel_for_array(iovs)?;
+        for iov in iovs.iter() {
+            let iov = memory.read(iov?)?;
+            if iov.buf_len == 0 {
+                continue;
+            }
+            let ret = iov.buf.as_array(iov.buf_len);
+            self.consume_fuel_for_array(ret)?;
+            return Ok(ret);
+        }
+        Ok(GuestPtr::new((0, 0)))
+    }
+
+    /// Copies the guest string `ptr` into the host.
+    ///
+    /// Consumes a corresponding amount of fuel for the byte size of `ptr` and
+    /// fails if it's too large.
+    fn read_string(&mut self, memory: &GuestMemory<'_>, ptr: GuestPtr<str>) -> Result<String> {
+        self.consume_fuel(usize::try_from(ptr.len())?)?;
+        Ok(memory.as_cow_str(ptr)?.into_owned())
+    }
 }
 
 impl WasiView for WasiP1Ctx {
@@ -558,6 +632,7 @@ impl WasiP1Ctx {
         ciovs: types::CiovecArray,
         write: FdWrite,
     ) -> Result<types::Size, types::Error> {
+        let buf = self.first_non_empty_ciovec(memory, ciovs)?;
         let t = self.transact()?;
         let desc = t.get_descriptor(fd)?;
         match desc {
@@ -579,7 +654,6 @@ impl WasiP1Ctx {
                 let append = *append;
                 drop(t);
                 let f = self.table.get(&fd)?.file()?;
-                let buf = first_non_empty_ciovec(memory, ciovs)?;
 
                 let do_write = move |f: &cap_std::fs::File, buf: &[u8]| match (append, write) {
                     // Note that this is implementing Linux semantics of
@@ -630,7 +704,6 @@ impl WasiP1Ctx {
                 }
                 let stream = stream.borrowed();
                 drop(t);
-                let buf = first_non_empty_ciovec(memory, ciovs)?;
                 let n = BlockingMode::Blocking
                     .write(memory, &mut self.table, stream, buf)
                     .await?
@@ -1124,46 +1197,14 @@ fn write_byte(memory: &mut GuestMemory<'_>, ptr: GuestPtr<u8>, byte: u8) -> Resu
     Ok(next)
 }
 
-fn read_string<'a>(memory: &'a GuestMemory<'_>, ptr: GuestPtr<str>) -> Result<String> {
-    Ok(memory.as_cow_str(ptr)?.into_owned())
-}
-
-// Returns the first non-empty buffer in `ciovs` or a single empty buffer if
-// they're all empty.
-fn first_non_empty_ciovec(
-    memory: &GuestMemory<'_>,
-    ciovs: types::CiovecArray,
-) -> Result<GuestPtr<[u8]>> {
-    for iov in ciovs.iter() {
-        let iov = memory.read(iov?)?;
-        if iov.buf_len == 0 {
-            continue;
-        }
-        return Ok(iov.buf.as_array(iov.buf_len));
-    }
-    Ok(GuestPtr::new((0, 0)))
-}
-
-// Returns the first non-empty buffer in `iovs` or a single empty buffer if
-// they're all empty.
-fn first_non_empty_iovec(
-    memory: &GuestMemory<'_>,
-    iovs: types::IovecArray,
-) -> Result<GuestPtr<[u8]>> {
-    for iov in iovs.iter() {
-        let iov = memory.read(iov?)?;
-        if iov.buf_len == 0 {
-            continue;
-        }
-        return Ok(iov.buf.as_array(iov.buf_len));
-    }
-    Ok(GuestPtr::new((0, 0)))
-}
-
 // Implement the WasiSnapshotPreview1 trait using only the traits that are
 // required for T, i.e., in terms of the preview 2 wit interface, and state
 // stored in the WasiP1Adapter struct.
 impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
+    fn set_hostcall_fuel(&mut self, fuel: usize) {
+        self.hostcall_fuel = fuel;
+    }
+
     #[instrument(skip(self, memory))]
     fn args_get(
         &mut self,
@@ -1646,6 +1687,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         fd: types::Fd,
         iovs: types::IovecArray,
     ) -> Result<types::Size, types::Error> {
+        let iov = self.first_non_empty_iovec(memory, iovs)?;
         let t = self.transact()?;
         let desc = t.get_descriptor(fd)?;
         match desc {
@@ -1662,7 +1704,6 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                 drop(t);
                 let pos = position.load(Ordering::Relaxed);
                 let file = self.table.get(&fd)?.file()?;
-                let iov = first_non_empty_iovec(memory, iovs)?;
                 let bytes_read = match (file.as_blocking_file(), memory.as_slice_mut(iov)?) {
                     // Try to read directly into wasm memory where possible
                     // when the current thread can block and additionally wasm
@@ -1701,15 +1742,14 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
             Descriptor::Stdin { stream, .. } => {
                 let stream = stream.borrowed();
                 drop(t);
-                let buf = first_non_empty_iovec(memory, iovs)?;
                 let read = BlockingMode::Blocking
-                    .read(&mut self.table, stream, buf.len().try_into()?)
+                    .read(&mut self.table, stream, iov.len().try_into()?)
                     .await?;
-                if read.len() > buf.len().try_into()? {
+                if read.len() > iov.len().try_into()? {
                     return Err(types::Errno::Range.into());
                 }
-                let buf = buf.get_range(0..u32::try_from(read.len())?).unwrap();
-                memory.copy_from_slice(&read, buf)?;
+                let iov = iov.get_range(0..u32::try_from(read.len())?).unwrap();
+                memory.copy_from_slice(&read, iov)?;
                 let n = read.len().try_into()?;
                 Ok(n)
             }
@@ -1727,6 +1767,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         iovs: types::IovecArray,
         offset: types::Filesize,
     ) -> Result<types::Size, types::Error> {
+        let buf = self.first_non_empty_iovec(memory, iovs)?;
         let t = self.transact()?;
         let desc = t.get_descriptor(fd)?;
         let (buf, read) = match desc {
@@ -1736,7 +1777,6 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                 let fd = fd.borrowed();
                 let blocking_mode = *blocking_mode;
                 drop(t);
-                let buf = first_non_empty_iovec(memory, iovs)?;
 
                 let stream = self.filesystem().read_via_stream(fd, offset)?;
                 let read = blocking_mode
@@ -2024,7 +2064,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         path: GuestPtr<str>,
     ) -> Result<(), types::Error> {
         let dirfd = self.get_dir_fd(dirfd)?;
-        let path = read_string(memory, path)?;
+        let path = self.read_string(memory, path)?;
         self.filesystem()
             .create_directory_at(dirfd.borrowed(), path)
             .await?;
@@ -2042,7 +2082,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         path: GuestPtr<str>,
     ) -> Result<types::Filestat, types::Error> {
         let dirfd = self.get_dir_fd(dirfd)?;
-        let path = read_string(memory, path)?;
+        let path = self.read_string(memory, path)?;
         let filesystem::DescriptorStat {
             type_,
             link_count: nlink,
@@ -2103,7 +2143,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         )?;
 
         let dirfd = self.get_dir_fd(dirfd)?;
-        let path = read_string(memory, path)?;
+        let path = self.read_string(memory, path)?;
         self.filesystem()
             .set_times_at(dirfd, flags.into(), path, atim, mtim)
             .await?;
@@ -2124,8 +2164,8 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
     ) -> Result<(), types::Error> {
         let src_fd = self.get_dir_fd(src_fd)?;
         let target_fd = self.get_dir_fd(target_fd)?;
-        let src_path = read_string(memory, src_path)?;
-        let target_path = read_string(memory, target_path)?;
+        let src_path = self.read_string(memory, src_path)?;
+        let target_path = self.read_string(memory, target_path)?;
         self.filesystem()
             .link_at(src_fd, src_flags.into(), src_path, target_fd, target_path)
             .await?;
@@ -2146,7 +2186,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         _fs_rights_inheriting: types::Rights,
         fdflags: types::Fdflags,
     ) -> Result<types::Fd, types::Error> {
-        let path = read_string(memory, path)?;
+        let path = self.read_string(memory, path)?;
 
         let mut flags = filesystem::DescriptorFlags::empty();
         if fs_rights_base.contains(types::Rights::FD_READ) {
@@ -2205,7 +2245,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         buf_len: types::Size,
     ) -> Result<types::Size, types::Error> {
         let dirfd = self.get_dir_fd(dirfd)?;
-        let path = read_string(memory, path)?;
+        let path = self.read_string(memory, path)?;
         let mut path = self
             .filesystem()
             .readlink_at(dirfd, path)
@@ -2228,7 +2268,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         path: GuestPtr<str>,
     ) -> Result<(), types::Error> {
         let dirfd = self.get_dir_fd(dirfd)?;
-        let path = read_string(memory, path)?;
+        let path = self.read_string(memory, path)?;
         self.filesystem().remove_directory_at(dirfd, path).await?;
         Ok(())
     }
@@ -2246,8 +2286,8 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
     ) -> Result<(), types::Error> {
         let src_fd = self.get_dir_fd(src_fd)?;
         let dest_fd = self.get_dir_fd(dest_fd)?;
-        let src_path = read_string(memory, src_path)?;
-        let dest_path = read_string(memory, dest_path)?;
+        let src_path = self.read_string(memory, src_path)?;
+        let dest_path = self.read_string(memory, dest_path)?;
         self.filesystem()
             .rename_at(src_fd, src_path, dest_fd, dest_path)
             .await?;
@@ -2263,8 +2303,8 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         dest_path: GuestPtr<str>,
     ) -> Result<(), types::Error> {
         let dirfd = self.get_dir_fd(dirfd)?;
-        let src_path = read_string(memory, src_path)?;
-        let dest_path = read_string(memory, dest_path)?;
+        let src_path = self.read_string(memory, src_path)?;
+        let dest_path = self.read_string(memory, dest_path)?;
         self.filesystem()
             .symlink_at(dirfd.borrowed(), src_path, dest_path)
             .await?;
@@ -2332,9 +2372,18 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
 
         let subs = subs.as_array(nsubscriptions);
         let events = events.as_array(nsubscriptions);
-
         let n = usize::try_from(nsubscriptions).unwrap_or(usize::MAX);
-        let mut pollables = Vec::with_capacity(n);
+
+        self.consume_fuel_for_array(subs)?;
+        self.consume_fuel_for_array(events)?;
+
+        let mut temp = TempResources {
+            ctx: self,
+            pollables: Vec::with_capacity(n),
+            inputs: Vec::new(),
+            outputs: Vec::new(),
+        };
+        let mut borrowed_pollables = Vec::with_capacity(n);
         for sub in subs.iter() {
             let sub = memory.read(sub?)?;
             let p = match sub.u {
@@ -2349,7 +2398,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                         types::Clockid::Monotonic => (timeout, absolute),
                         types::Clockid::Realtime if !absolute => (timeout, false),
                         types::Clockid::Realtime => {
-                            let now = wall_clock::Host::now(&mut self.clocks())
+                            let now = wall_clock::Host::now(&mut temp.ctx.clocks())
                                 .context("failed to call `wall_clock::now`")
                                 .map_err(types::Error::trap)?;
 
@@ -2372,11 +2421,11 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                         _ => return Err(types::Errno::Inval.into()),
                     };
                     if absolute {
-                        monotonic_clock::Host::subscribe_instant(&mut self.clocks(), timeout)
+                        monotonic_clock::Host::subscribe_instant(&mut temp.ctx.clocks(), timeout)
                             .context("failed to call `monotonic_clock::subscribe_instant`")
                             .map_err(types::Error::trap)?
                     } else {
-                        monotonic_clock::Host::subscribe_duration(&mut self.clocks(), timeout)
+                        monotonic_clock::Host::subscribe_duration(&mut temp.ctx.clocks(), timeout)
                             .context("failed to call `monotonic_clock::subscribe_duration`")
                             .map_err(types::Error::trap)?
                     }
@@ -2385,29 +2434,32 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                     file_descriptor,
                 }) => {
                     let stream = {
-                        let t = self.transact()?;
+                        let t = temp.ctx.transact()?;
                         let desc = t.get_descriptor(file_descriptor)?;
                         match desc {
                             Descriptor::Stdin { stream, .. } => stream.borrowed(),
                             Descriptor::File(File { fd, position, .. }) => {
                                 let pos = position.load(Ordering::Relaxed);
                                 let fd = fd.borrowed();
                                 drop(t);
-                                self.filesystem().read_via_stream(fd, pos)?
+                                let r = temp.ctx.filesystem().read_via_stream(fd, pos)?;
+                                let ret = r.borrowed();
+                                temp.inputs.push(r);
+                                ret
                             }
                             // TODO: Support sockets
                             _ => return Err(types::Errno::Badf.into()),
                         }
                     };
-                    streams::HostInputStream::subscribe(&mut self.table, stream)
+                    streams::HostInputStream::subscribe(&mut temp.ctx.table, stream)
                         .context("failed to call `subscribe` on `input-stream`")
                         .map_err(types::Error::trap)?
                 }
                 types::SubscriptionU::FdWrite(types::SubscriptionFdReadwrite {
                     file_descriptor,
                 }) => {
                     let stream = {
-                        let t = self.transact()?;
+                        let t = temp.ctx.transact()?;
                         let desc = t.get_descriptor(file_descriptor)?;
                         match desc {
                             Descriptor::Stdout { stream, .. }
@@ -2422,32 +2474,38 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                                 let position = position.clone();
                                 let append = *append;
                                 drop(t);
-                                if append {
-                                    self.filesystem().append_via_stream(fd)?
+                                let r = if append {
+                                    temp.ctx.filesystem().append_via_stream(fd)?
                                 } else {
                                     let pos = position.load(Ordering::Relaxed);
-                                    self.filesystem().write_via_stream(fd, pos)?
-                                }
+                                    temp.ctx.filesystem().write_via_stream(fd, pos)?
+                                };
+                                let ret = r.borrowed();
+                                temp.outputs.push(r);
+                                ret
                             }
                             // TODO: Support sockets
                             _ => return Err(types::Errno::Badf.into()),
                         }
                     };
-                    streams::HostOutputStream::subscribe(&mut self.table, stream)
+                    streams::HostOutputStream::subscribe(&mut temp.ctx.table, stream)
                         .context("failed to call `subscribe` on `output-stream`")
                         .map_err(types::Error::trap)?
                 }
             };
-            pollables.push(p);
+            borrowed_pollables.push(p.borrowed());
+            temp.pollables.push(p);
         }
-        let ready: HashSet<_> = self
+        let ready: HashSet<_> = temp
+            .ctx
             .table
-            .poll(pollables)
+            .poll(borrowed_pollables)
             .await
             .context("failed to call `poll-oneoff`")
             .map_err(types::Error::trap)?
             .into_iter()
             .collect();
+        drop(temp);
 
         let mut count: types::Size = 0;
         for (sub, event) in (0..)
@@ -2544,7 +2602,31 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                 .checked_add(1)
                 .ok_or_else(|| types::Error::from(types::Errno::Overflow))?
         }
-        Ok(count)
+        return Ok(count);
+
+        struct TempResources<'a> {
+            ctx: &'a mut WasiP1Ctx,
+            pollables: Vec<Resource<crate::p2::bindings::io::streams::Pollable>>,
+            inputs: Vec<Resource<crate::p2::bindings::io::streams::InputStream>>,
+            outputs: Vec<Resource<crate::p2::bindings::io::streams::OutputStream>>,
+        }
+
+        impl Drop for TempResources<'_> {
+            fn drop(&mut self) {
+                for p in self.pollables.drain(..) {
+                    use wasmtime_wasi_io::bindings::wasi::io::poll::HostPollable;
+                    self.ctx.table.drop(p).unwrap();
+                }
+                for p in self.inputs.drain(..) {
+                    assert!(p.owned());
+                    self.ctx.table.delete(p).unwrap();
+                }
+                for p in self.outputs.drain(..) {
+                    assert!(p.owned());
+                    self.ctx.table.delete(p).unwrap();
+                }
+            }
+        }
     }
 
     #[instrument(skip(self, _memory))]

@@ -86,7 +86,7 @@ impl FileInputStream {
     fn blocking_read(file: &cap_std::fs::File, offset: u64, size: usize) -> ReadState {
         use system_interface::fs::FileIoExt;
 
-        let mut buf = BytesMut::zeroed(size);
+        let mut buf = BytesMut::zeroed(size.min(crate::MAX_READ_SIZE_ALLOC));
         loop {
             match file.read_at(&mut buf, offset) {
                 Ok(0) => return ReadState::Closed,

@@ -115,7 +115,12 @@ impl HostDescriptor for WasiFilesystemCtxView<'_> {
 
         let (mut buffer, r) = f
             .run_blocking(move |f| {
-                let mut buffer = vec![0; len.try_into().unwrap_or(usize::MAX)];
+                let mut buffer = vec![
+                    0;
+                    len.try_into()
+                        .unwrap_or(usize::MAX)
+                        .min(crate::MAX_READ_SIZE_ALLOC)
+                ];
                 let r = f.read_vectored_at(&mut [IoSliceMut::new(&mut buffer)], offset);
                 (buffer, r)
             })

@@ -1,9 +1,13 @@
 use crate::p2::bindings::random::{insecure, insecure_seed, random};
 use crate::random::WasiRandomCtx;
+use anyhow::bail;
 use cap_rand::{Rng, distributions::Standard};
 
 impl random::Host for WasiRandomCtx {
     fn get_random_bytes(&mut self, len: u64) -> anyhow::Result<Vec<u8>> {
+        if len > self.max_size {
+            bail!("requested len {len:?} exceeds limit {}", self.max_size);
+        }
         Ok((&mut self.random)
             .sample_iter(Standard)
             .take(len as usize)
@@ -17,6 +21,9 @@ impl random::Host for WasiRandomCtx {
 
 impl insecure::Host for WasiRandomCtx {
     fn get_insecure_random_bytes(&mut self, len: u64) -> anyhow::Result<Vec<u8>> {
+        if len > self.max_size {
+            bail!("requested len {len:?} exceeds limit {}", self.max_size);
+        }
         Ok((&mut self.insecure_random)
             .sample_iter(Standard)
             .take(len as usize)

@@ -216,10 +216,12 @@ impl udp::HostIncomingDatagramStream for WasiSocketsCtxView<'_> {
         }
 
         let mut datagrams = vec![];
+        let mut sum = 0;
 
-        while datagrams.len() < max_results {
+        while datagrams.len() < max_results && sum < crate::MAX_READ_SIZE_ALLOC {
             match recv_one(stream) {
                 Ok(Some(datagram)) => {
+                    sum += 1 + datagram.data.len();
                     datagrams.push(datagram);
                 }
                 Ok(None) => {

@@ -163,7 +163,7 @@ impl AsyncReadStream {
             let mut reader = pin!(reader);
             loop {
                 use tokio::io::AsyncReadExt;
-                let mut buf = bytes::BytesMut::with_capacity(4096);
+                let mut buf = bytes::BytesMut::with_capacity(crate::MAX_READ_SIZE_ALLOC);
                 let sent = match reader.read_buf(&mut buf).await {
                     Ok(nbytes) if nbytes == 0 => sender.send(Err(StreamError::Closed)).await,
                     Ok(_) => sender.send(Ok(buf.freeze())).await,
@@ -516,29 +516,31 @@ mod test {
     // suitable design for all applications, and we will probably make a knob or change the
     // behavior at some point, but this test shows the behavior as it is implemented:
     async fn backpressure_read_stream() {
-        let (r, mut w) = simplex(16 * 1024); // Make sure this buffer isn't a bottleneck
+        let (r, mut w) = simplex(4 * crate::MAX_READ_SIZE_ALLOC); // Make sure this buffer isn't a bottleneck
         let mut reader = AsyncReadStream::new(r);
 
         let writer_task = tokio::task::spawn(async move {
             // Write twice as much as we can buffer up in an AsyncReadStream:
-            w.write_all(&[123; 8192]).await.unwrap();
+            w.write_all(&[123; 2 * crate::MAX_READ_SIZE_ALLOC])
+                .await
+                .unwrap();
             w
         });
 
         resolves_immediately(reader.ready()).await;
 
         // Now we expect the reader task has sent 4k from the stream to the reader.
         // Try to read out one bigger than the buffer available:
-        let bs = reader.read(4097).unwrap();
-        assert_eq!(bs.len(), 4096);
+        let bs = reader.read(crate::MAX_READ_SIZE_ALLOC + 1).unwrap();
+        assert_eq!(bs.len(), crate::MAX_READ_SIZE_ALLOC);
 
         // Allow the crank to turn more:
         resolves_immediately(reader.ready()).await;
 
         // Again we expect the reader task has sent 4k from the stream to the reader.
         // Try to read out one bigger than the buffer available:
-        let bs = reader.read(4097).unwrap();
-        assert_eq!(bs.len(), 4096);
+        let bs = reader.read(crate::MAX_READ_SIZE_ALLOC + 1).unwrap();
+        assert_eq!(bs.len(), crate::MAX_READ_SIZE_ALLOC);
 
         // The writer task is now finished - join with it:
         let w = resolves_immediately(writer_task).await;
@@ -550,7 +552,10 @@ mod test {
         resolves_immediately(reader.ready()).await;
 
         // Now we expect the reader to be empty, and the stream.dropd:
-        assert!(matches!(reader.read(4097), Err(StreamError::Closed)));
+        assert!(matches!(
+            reader.read(crate::MAX_READ_SIZE_ALLOC + 1),
+            Err(StreamError::Closed)
+        ));
     }
 
     #[test_log::test(test_log::test(tokio::test(flavor = "multi_thread")))]

@@ -69,7 +69,7 @@ impl TcpReader {
             return Ok(bytes::Bytes::new());
         }
 
-        let mut buf = bytes::BytesMut::with_capacity(size);
+        let mut buf = bytes::BytesMut::with_capacity(size.min(crate::MAX_READ_SIZE_ALLOC));
         let n = match self.stream.try_read_buf(&mut buf) {
             // A 0-byte read indicates that the stream has closed.
             Ok(0) => {

@@ -41,10 +41,15 @@ impl HasData for WasiRandom {
     type Data<'a> = &'a mut WasiRandomCtx;
 }
 
+/// Default largest length accepted by wasi 0.2 `get-random-bytes` and
+/// `get-insecure-random-bytes` methods.
+pub const DEFAULT_MAX_SIZE: u64 = 2 << 30;
+
 pub struct WasiRandomCtx {
     pub(crate) random: Box<dyn RngCore + Send>,
     pub(crate) insecure_random: Box<dyn RngCore + Send>,
     pub(crate) insecure_random_seed: u128,
+    pub(crate) max_size: u64,
 }
 
 impl Default for WasiRandomCtx {
@@ -60,10 +65,12 @@ impl Default for WasiRandomCtx {
         // API.
         let insecure_random_seed =
             cap_rand::thread_rng(cap_rand::ambient_authority()).r#gen::<u128>();
+        let max_size = DEFAULT_MAX_SIZE;
         Self {
             random: thread_rng(),
             insecure_random,
             insecure_random_seed,
+            max_size,
         }
     }
 }

@@ -3,6 +3,7 @@ use anyhow::Result;
 use std::path::Path;
 use test_programs_artifacts::*;
 use wasmtime::{Linker, Module};
+use wasmtime_wasi::WasiView;
 use wasmtime_wasi::p1::{WasiP1Ctx, add_to_linker_async};
 
 async fn run(path: &str, inherit_stdio: bool) -> Result<()> {
@@ -21,6 +22,7 @@ async fn run(path: &str, inherit_stdio: bool) -> Result<()> {
         }
         builder.build_p1()
     })?;
+    store.data_mut().wasi.ctx().table.set_max_capacity(1000);
     let instance = linker.instantiate_async(&mut store, &module).await?;
     let start = instance.get_typed_func::<(), ()>(&mut store, "_start")?;
     start.call_async(&mut store, ()).await?;
@@ -243,9 +245,7 @@ async fn p1_file_write() {
 async fn p1_path_open_lots() {
     run(P1_PATH_OPEN_LOTS, true).await.unwrap()
 }
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p1_cli_much_stdout() {}
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
+async fn p1_sleep_quickly_but_lots() {
+    run(P1_SLEEP_QUICKLY_BUT_LOTS, true).await.unwrap()
+}

@@ -19,10 +19,7 @@ async fn run(path: &str, inherit_stdio: bool) -> Result<()> {
         if inherit_stdio {
             builder.inherit_stdio();
         }
-        MyWasiCtx {
-            wasi: builder.build(),
-            table: Default::default(),
-        }
+        MyWasiCtx::new(builder.build())
     })?;
     let component = Component::from_file(&engine, path)?;
     let command = Command::instantiate_async(&mut store, &component, &linker).await?;
@@ -262,6 +259,12 @@ async fn p1_file_write() {
 async fn p1_path_open_lots() {
     run(P1_PATH_OPEN_LOTS_COMPONENT, false).await.unwrap()
 }
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
+async fn p1_sleep_quickly_but_lots() {
+    run(P1_SLEEP_QUICKLY_BUT_LOTS_COMPONENT, false)
+        .await
+        .unwrap()
+}
 
 #[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn p2_sleep() {
@@ -359,8 +362,3 @@ async fn p2_adapter_badfd() {
 async fn p2_file_read_write() {
     run(P2_FILE_READ_WRITE_COMPONENT, false).await.unwrap()
 }
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p1_cli_much_stdout() {}

@@ -21,10 +21,7 @@ fn run(path: &str, inherit_stdio: bool) -> Result<()> {
                 builder.inherit_stdio();
             }
             builder.allow_blocking_current_thread(blocking);
-            MyWasiCtx {
-                wasi: builder.build(),
-                table: Default::default(),
-            }
+            MyWasiCtx::new(builder.build())
         })?;
         let command = Command::instantiate(&mut store, &component, &linker)?;
         command
@@ -250,6 +247,10 @@ fn p1_file_write() {
 fn p1_path_open_lots() {
     run(P1_PATH_OPEN_LOTS_COMPONENT, false).unwrap()
 }
+#[test_log::test]
+fn p1_sleep_quickly_but_lots() {
+    run(P1_SLEEP_QUICKLY_BUT_LOTS_COMPONENT, false).unwrap()
+}
 
 #[test_log::test]
 fn p2_sleep() {
@@ -339,8 +340,3 @@ fn p2_adapter_badfd() {
 fn p2_file_read_write() {
     run(P2_FILE_READ_WRITE_COMPONENT, false).unwrap()
 }
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p1_cli_much_stdout() {}

@@ -26,11 +26,12 @@ async fn run_allow_blocking_current_thread(
         .context("failed to link `wasi:cli@0.2.x`")?;
     wasmtime_wasi::p3::add_to_linker(&mut linker).context("failed to link `wasi:cli@0.3.x`")?;
 
-    let (mut store, _td) = Ctx::new(&engine, name, |builder| MyWasiCtx {
-        wasi: builder
-            .allow_blocking_current_thread(allow_blocking_current_thread)
-            .build(),
-        table: Default::default(),
+    let (mut store, _td) = Ctx::new(&engine, name, |builder| {
+        MyWasiCtx::new(
+            builder
+                .allow_blocking_current_thread(allow_blocking_current_thread)
+                .build(),
+        )
     })?;
     let component = Component::from_file(&engine, path)?;
     let command = Command::instantiate_async(&mut store, &component, &linker)
@@ -150,27 +151,3 @@ async fn p3_file_write() -> anyhow::Result<()> {
 async fn p3_file_write_blocking() -> anyhow::Result<()> {
     run_allow_blocking_current_thread(P3_FILE_WRITE_COMPONENT, true).await
 }
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p3_cli_hello_stdout() {}
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p3_cli_much_stdout() {}
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p3_cli_serve_hello_world() {}
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p3_cli_serve_sleep() {}

@@ -72,8 +72,16 @@ impl<T> Drop for Ctx<T> {
 }
 
 pub struct MyWasiCtx {
-    pub wasi: WasiCtx,
-    pub table: ResourceTable,
+    wasi: WasiCtx,
+    table: ResourceTable,
+}
+
+impl MyWasiCtx {
+    pub fn new(wasi: WasiCtx) -> Self {
+        let mut table = ResourceTable::new();
+        table.set_max_capacity(1000);
+        Self { wasi, table }
+    }
 }
 
 impl WasiView for Ctx<MyWasiCtx> {

@@ -5024,13 +5024,11 @@ impl TaskId {
     /// and delete the task when all threads are done.
     pub(crate) fn host_future_dropped<T>(&self, store: StoreContextMut<T>) -> Result<()> {
         let task = store.0.concurrent_state_mut().get_mut(self.task)?;
-        if !task.already_lowered_parameters() {
-            Waitable::Guest(self.task).delete_from(store.0.concurrent_state_mut())?
-        } else {
+        if task.already_lowered_parameters() {
             task.host_future_state = HostFutureState::Dropped;
-            if task.ready_to_delete() {
-                Waitable::Guest(self.task).delete_from(store.0.concurrent_state_mut())?
-            }
+        }
+        if task.ready_to_delete() {
+            Waitable::Guest(self.task).delete_from(store.0.concurrent_state_mut())?
         }
         Ok(())
     }
@@ -5073,8 +5071,6 @@ pub(crate) fn prepare_call<T, R>(
     let token = StoreToken::new(store.as_context_mut());
     let state = store.0.concurrent_state_mut();
 
-    assert!(state.guest_thread.is_none());
-
     let (tx, rx) = oneshot::channel();
     let (exit_tx, exit_rx) = oneshot::channel();
 

@@ -334,6 +334,12 @@ pub struct LiftContext<'a> {
         allow(unused, reason = "easier to not #[cfg] away")
     )]
     concurrent_state: &'a mut ConcurrentState,
+
+    /// Remaining fuel for this hostcall/lift operation.
+    ///
+    /// This is decremented for strings/lists, for example, to cap the size of
+    /// data the host allocates on behalf of the guest.
+    hostcall_fuel: usize,
 }
 
 #[doc(hidden)]
@@ -346,6 +352,7 @@ impl<'a> LiftContext<'a> {
         instance_handle: Instance,
     ) -> LiftContext<'a> {
         let store_id = store.id();
+        let hostcall_fuel = store.hostcall_fuel();
         // From `&mut StoreOpaque` provided the goal here is to project out
         // three different disjoint fields owned by the store: memory,
         // `CallContexts`, and `HandleTable`. There's no native API for that
@@ -369,6 +376,7 @@ impl<'a> LiftContext<'a> {
             host_table,
             host_resource_data,
             concurrent_state,
+            hostcall_fuel,
         }
     }
 
@@ -490,4 +498,21 @@ impl<'a> LiftContext<'a> {
     pub fn exit_call(&mut self) -> Result<()> {
         self.resource_tables().exit_call()
     }
+
+    /// Consumes `amt` units of fuel, typically a number of bytes, from this
+    /// context.
+    ///
+    /// Returns an error if the fuel is exhausted which will cause a trap in the
+    /// guest. Note that this is distinct from Wasm's fuel, this is just for
+    /// keeping track of data flowing from the guest to the host.
+    pub fn consume_fuel(&mut self, amt: usize) -> Result<()> {
+        match self.hostcall_fuel.checked_sub(amt) {
+            Some(new) => self.hostcall_fuel = new,
+            None => bail!(
+                "too much data is being copied between the host and the guest: \
+                 fuel allocated for hostcalls has been exhausted"
+            ),
+        }
+        Ok(())
+    }
 }

@@ -1637,7 +1637,7 @@ impl WasmStr {
             }
         };
         match byte_len.and_then(|len| ptr.checked_add(len)) {
-            Some(n) if n <= cx.memory().len() => {}
+            Some(n) if n <= cx.memory().len() => cx.consume_fuel(n - ptr)?,
             _ => bail!("string pointer/length out of bounds of memory"),
         }
         Ok(WasmStr {
@@ -1895,7 +1895,7 @@ impl<T: Lift> WasmList<T> {
             .checked_mul(T::SIZE32)
             .and_then(|len| ptr.checked_add(len))
         {
-            Some(n) if n <= cx.memory().len() => {}
+            Some(n) if n <= cx.memory().len() => cx.consume_fuel(n - ptr)?,
             _ => bail!("list pointer/length out of bounds of memory"),
         }
         if ptr % usize::try_from(T::ALIGN32)? != 0 {

@@ -36,6 +36,7 @@ impl core::error::Error for ResourceTableError {}
 pub struct ResourceTable {
     entries: Vec<Entry>,
     free_head: Option<usize>,
+    max_capacity: usize,
 }
 
 #[derive(Debug)]
@@ -66,6 +67,11 @@ struct Tombstone;
 // necessary.
 const DELETE_WITH_TOMBSTONE: bool = false;
 
+/// Default setting for `ResourceTable::max_capacity`, chosen to be high
+/// enough that it doesn't need changing all that often but low enough that
+/// exhausting it isn't a massive problem for the host.
+const DEFAULT_MAX_CAPACITY: usize = 1_000_000;
+
 /// This structure tracks parent and child relationships for a given table entry.
 ///
 /// Parents and children are referred to by table index. We maintain the
@@ -105,10 +111,7 @@ impl TableEntry {
 impl ResourceTable {
     /// Create an empty table
     pub fn new() -> Self {
-        ResourceTable {
-            entries: Vec::new(),
-            free_head: None,
-        }
+        ResourceTable::with_capacity(0)
     }
 
     /// Returns whether or not this table is empty.
@@ -122,11 +125,28 @@ impl ResourceTable {
         })
     }
 
+    /// Returns the maximum capacity of this table, in elements, before adding
+    /// any more will be refused.
+    pub fn max_capacity(&self) -> usize {
+        self.max_capacity
+    }
+
+    /// Configures the maximum number of entries that may be present within this
+    /// table.
+    ///
+    /// Note that this does not retroactively shrink the table nor evict
+    /// existing entries should the maximum be smaller than the current size of
+    /// the entry table.
+    pub fn set_max_capacity(&mut self, max: usize) {
+        self.max_capacity = max;
+    }
+
     /// Create an empty table with at least the specified capacity.
     pub fn with_capacity(capacity: usize) -> Self {
         ResourceTable {
             entries: Vec::with_capacity(capacity),
             free_head: None,
+            max_capacity: DEFAULT_MAX_CAPACITY,
         }
     }
 
@@ -197,6 +217,9 @@ impl ResourceTable {
             self.entries[free] = Entry::Occupied { entry: e };
             Ok(free.try_into().unwrap())
         } else {
+            if self.entries.len() >= self.max_capacity {
+                return Err(ResourceTableError::Full);
+            }
             let ix = self
                 .entries
                 .len()
@@ -446,3 +469,34 @@ pub fn test_free_list() {
     let x = table.push(()).unwrap();
     assert_eq!(x.rep(), 2);
 }
+
+#[test]
+fn test_max_capacity() {
+    let mut table = ResourceTable::new();
+    assert_eq!(table.max_capacity(), DEFAULT_MAX_CAPACITY);
+
+    table.set_max_capacity(0);
+    assert_eq!(table.max_capacity(), 0);
+    assert!(table.push(()).is_err());
+
+    table.set_max_capacity(1);
+    assert_eq!(table.max_capacity(), 1);
+    let x = table.push(()).unwrap();
+    assert!(table.push(()).is_err());
+
+    table.set_max_capacity(0);
+    assert!(table.push(()).is_err());
+    table.delete(x).unwrap();
+    let x = table.push(()).unwrap();
+    table.delete(x).unwrap();
+
+    table.set_max_capacity(10);
+
+    let handles = (0..10).map(|_| table.push(()).unwrap()).collect::<Vec<_>>();
+    assert!(table.push(()).is_err());
+    for handle in handles {
+        table.delete(handle).unwrap();
+    }
+
+    table.push(()).unwrap();
+}

@@ -2,14 +2,38 @@
 use crate::runtime::vm::VMStore;
 use crate::runtime::vm::component::{ComponentInstance, OwnedComponentInstance};
 use crate::store::{StoreData, StoreId, StoreOpaque};
+use crate::{AsContext, AsContextMut, Store, StoreContextMut};
 #[cfg(feature = "component-model-async")]
 use alloc::vec::Vec;
 use core::pin::Pin;
 use wasmtime_environ::PrimaryMap;
 
-#[derive(Default)]
+/// Default amount of fuel allowed for all guest-to-host calls in the component
+/// model.
+///
+/// This is the maximal amount of data which will be copied from the guest to
+/// the host by default. This is set to a very large value to avoid breaking
+/// existing embeddings for when this feature was backported.
+const DEFAULT_HOSTCALL_FUEL: usize = 2 << 30;
+
 pub struct ComponentStoreData {
     instances: PrimaryMap<ComponentInstanceId, Option<OwnedComponentInstance>>,
+
+    /// Fuel to be used for each time the guest calls the host or transfers data
+    /// to the host.
+    ///
+    /// Caps the size of the allocations made on the host to this amount
+    /// effectively.
+    hostcall_fuel: usize,
+}
+
+impl Default for ComponentStoreData {
+    fn default() -> Self {
+        Self {
+            instances: PrimaryMap::default(),
+            hostcall_fuel: DEFAULT_HOSTCALL_FUEL,
+        }
+    }
 }
 
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
@@ -84,6 +108,14 @@ impl StoreOpaque {
     pub(crate) fn component_instance(&self, id: ComponentInstanceId) -> &ComponentInstance {
         self.store_data().component_instance(id)
     }
+
+    pub(crate) fn hostcall_fuel(&self) -> usize {
+        self.store_data().components.hostcall_fuel
+    }
+
+    pub(crate) fn set_hostcall_fuel(&mut self, fuel: usize) {
+        self.store_data_mut().components.hostcall_fuel = fuel;
+    }
 }
 
 /// A type used to represent an allocated `ComponentInstance` located within a
@@ -173,3 +205,55 @@ impl StoreComponentInstanceId {
         store.component_instance_mut(self.instance)
     }
 }
+
+impl<T> Store<T> {
+    /// Returns the amount of "hostcall fuel" used for guest-to-host component
+    /// calls.
+    ///
+    /// This is either the default amount if it hasn't been configured or
+    /// returns the last value passed to [`Store::set_hostcall_fuel`].
+    ///
+    /// See [`Store::set_hostcall_fuel`] `for more details.
+    pub fn hostcall_fuel(&self) -> usize {
+        self.as_context().0.hostcall_fuel()
+    }
+
+    /// Sets the amount of "hostcall fuel" used for guest-to-host component
+    /// calls.
+    ///
+    /// Whenever the guest calls the host it often wants to transfer some data
+    /// as well, such as strings or lists. This configured fuel value can be
+    /// used to limit the amount of data that the host allocates on behalf of
+    /// the guest. This is a DoS mitigation mechanism to prevent a malicious
+    /// guest from causing the host to allocate an unbounded amount of memory
+    /// for example.
+    ///
+    /// Fuel is considered distinct for each host call. The host is responsible
+    /// for ensuring it retains a proper amount of data between host calls if
+    /// applicable. The `fuel` provided here will be the initial value for each
+    /// time the guest calls the host.
+    ///
+    /// The `fuel` value here should roughly corresponds to the maximal number
+    /// of bytes that the guest may transfer to the host in one call.
+    ///
+    /// Note that data transferred from the host to the guest is not limited
+    /// because it's already resident on the host itself. Only data from the
+    /// guest to the host is limited.
+    ///
+    /// The default value for this is 128 MiB.
+    pub fn set_hostcall_fuel(&mut self, fuel: usize) {
+        self.as_context_mut().set_hostcall_fuel(fuel)
+    }
+}
+
+impl<T> StoreContextMut<'_, T> {
+    /// See [`Store::hostcall_fuel`].
+    pub fn hostcall_fuel(&self) -> usize {
+        self.0.hostcall_fuel()
+    }
+
+    /// See [`Store::set_hostcall_fuel`].
+    pub fn set_hostcall_fuel(&mut self, fuel: usize) {
+        self.0.set_hostcall_fuel(fuel)
+    }
+}

@@ -939,7 +939,7 @@ fn load_list(cx: &mut LiftContext<'_>, ty: TypeListIndex, ptr: usize, len: usize
         .checked_mul(element_size)
         .and_then(|len| ptr.checked_add(len))
     {
-        Some(n) if n <= cx.memory().len() => {}
+        Some(n) if n <= cx.memory().len() => cx.consume_fuel(n - ptr)?,
         _ => bail!("list pointer/length out of bounds of memory"),
     }
     if ptr % usize::try_from(element_alignment)? != 0 {

@@ -81,6 +81,13 @@ pub fn define_module_trait(m: &Module, settings: &CodegenSettings) -> TokenStrea
     quote! {
         pub trait #traitname {
             #(#traitmethods)*
+
+            /// Indicates, if this implementation supports it, that the
+            /// specified amount of fuel should be the maximal spent for the
+            /// upcoming function call.
+            fn set_hostcall_fuel(&mut self, fuel: usize) {
+                let _ = fuel;
+            }
         }
     }
 }

@@ -116,14 +116,17 @@ fn generate_func(
 
     let body = quote! {
         let export = caller.get_export("memory");
+        let fuel = wiggle::wasmtime_crate::AsContextMut::as_context_mut(&mut caller).hostcall_fuel();
         let (mut mem, ctx) = match &export {
             Some(wiggle::wasmtime_crate::Extern::Memory(m)) => {
                 let (mem, ctx) = m.data_and_store_mut(&mut caller);
                 let ctx = get_cx(ctx);
+                ctx.set_hostcall_fuel(fuel);
                 (wiggle::GuestMemory::Unshared(mem), ctx)
             }
             Some(wiggle::wasmtime_crate::Extern::SharedMemory(m)) => {
                 let ctx = get_cx(caller.data_mut());
+                ctx.set_hostcall_fuel(fuel);
                 (wiggle::GuestMemory::Shared(m.data()), ctx)
             }
             _ => wiggle::anyhow::bail!("missing required memory export"),

@@ -1,3 +1,32 @@
+## 40.0.4
+
+Released 2026-02-24.
+
+### Changed
+
+* Wasmtime's implementation of WASI now has the ability to limit resource
+  consumption on behalf of the guest, such as host-allocated memory. This means
+  that some behaviors previously allowed by Wasmtime can now disallowed, such as
+  transferring excessive data from the guest to the host. Additionally calls to
+  `wasi:random/random.get-random-bytes`, for example, can have limits in place
+  to avoid allocating too much memory on the host. To preserve
+  backwards-compatible behavior these limits are NOT set by default. Embedders
+  must opt-in to configuring these knobs as appropriate for their embeddings.
+  For more information on this see the related security advisory with further
+  details on knobs added and what behaviors can be restricted.
+  [GHSA-852m-cvvp-9p4w](https://github.com/bytecodealliance/wasmtime/security/advisories/GHSA-852m-cvvp-9p4w)
+
+### Fixed
+
+* Panics when adding too many headers to a `wasi:http/types.fields` has been
+  resolved
+  [GHSA-243v-98vx-264h](https://github.com/bytecodealliance/wasmtime/security/advisories/GHSA-243v-98vx-264h)
+
+* Panic when dropping a `[Typed]Func::call_async` future.
+  [GHSA-xjhv-v822-pf94](https://github.com/bytecodealliance/wasmtime/security/advisories/GHSA-xjhv-v822-pf94)
+
+--------------------------------------------------------------------------------
+
 ## 40.0.3
 
 Released 2026-01-26.

@@ -1035,8 +1035,8 @@ impl RunCommand {
                         }
                     }
                 }
-
-                store.data_mut().wasi_http = Some(Arc::new(WasiHttpCtx::new()));
+                let http = self.run.wasi_http_ctx()?;
+                store.data_mut().wasi_http = Some(Arc::new(http));
             }
         }
 
@@ -1131,7 +1131,13 @@ impl RunCommand {
         let mut builder = wasmtime_wasi::WasiCtxBuilder::new();
         builder.inherit_stdio().args(&self.compute_argv()?);
         self.run.configure_wasip2(&mut builder)?;
-        let ctx = builder.build_p1();
+        let mut ctx = builder.build_p1();
+        if let Some(max) = self.run.common.wasi.max_resources {
+            ctx.ctx().table.set_max_capacity(max);
+        }
+        if let Some(fuel) = self.run.common.wasi.hostcall_fuel {
+            store.set_hostcall_fuel(fuel);
+        }
         store.data_mut().wasip1_ctx = Some(Arc::new(Mutex::new(ctx)));
         Ok(())
     }

@@ -230,10 +230,14 @@ impl ServeCommand {
         builder.stdout(LogStream::new(stdout_prefix, Output::Stdout));
         builder.stderr(LogStream::new(stderr_prefix, Output::Stderr));
 
+        let mut table = wasmtime::component::ResourceTable::new();
+        if let Some(max) = self.run.common.wasi.max_resources {
+            table.set_max_capacity(max);
+        }
         let mut host = Host {
-            table: wasmtime::component::ResourceTable::new(),
+            table,
             ctx: builder.build(),
-            http: WasiHttpCtx::new(),
+            http: self.run.wasi_http_ctx()?,
             http_outgoing_body_buffer_chunks: self.run.common.wasi.http_outgoing_body_buffer_chunks,
             http_outgoing_body_chunk_size: self.run.common.wasi.http_outgoing_body_chunk_size,
 
@@ -301,6 +305,10 @@ impl ServeCommand {
 
         let mut store = Store::new(engine, host);
 
+        if let Some(fuel) = self.run.common.wasi.hostcall_fuel {
+            store.set_hostcall_fuel(fuel);
+        }
+
         store.data_mut().limits = self.run.store_limits();
         store.limiter(|t| &mut t.limits);
 

@@ -323,10 +323,22 @@ impl RunCommon {
         if let Some(enable) = self.common.wasi.udp {
             builder.allow_udp(enable);
         }
+        if let Some(max_size) = self.common.wasi.max_random_size {
+            builder.max_random_size(max_size);
+        }
 
         Ok(())
     }
 
+    #[cfg(feature = "wasi-http")]
+    pub fn wasi_http_ctx(&self) -> Result<wasmtime_wasi_http::WasiHttpCtx> {
+        let mut http = wasmtime_wasi_http::WasiHttpCtx::new();
+        if let Some(limit) = self.common.wasi.max_http_fields_size {
+            http.set_field_size_limit(limit);
+        }
+        Ok(http)
+    }
+
     pub fn compute_preopen_sockets(&self) -> Result<Vec<TcpListener>> {
         let mut listeners = vec![];
 

@@ -172,9 +172,12 @@ fn assert_trap_code(status: &ExitStatus) {
 // Run a simple WASI hello world, snapshot0 edition.
 #[test]
 fn hello_wasi_snapshot0() -> Result<()> {
-    let wasm = build_wasm("tests/all/cli_tests/hello_wasi_snapshot0.wat")?;
     for preview2 in ["-Spreview2=n", "-Spreview2=y"] {
-        let stdout = run_wasmtime(&["-Ccache=n", preview2, wasm.path().to_str().unwrap()])?;
+        let stdout = run_wasmtime(&[
+            "-Ccache=n",
+            preview2,
+            "tests/all/cli_tests/hello_wasi_snapshot0.wat",
+        ])?;
         assert_eq!(stdout, "Hello, world!\n");
     }
     Ok(())
@@ -183,8 +186,7 @@ fn hello_wasi_snapshot0() -> Result<()> {
 // Run a simple WASI hello world, snapshot1 edition.
 #[test]
 fn hello_wasi_snapshot1() -> Result<()> {
-    let wasm = build_wasm("tests/all/cli_tests/hello_wasi_snapshot1.wat")?;
-    let stdout = run_wasmtime(&["-Ccache=n", wasm.path().to_str().unwrap()])?;
+    let stdout = run_wasmtime(&["-Ccache=n", "tests/all/cli_tests/hello_wasi_snapshot1.wat"])?;
     assert_eq!(stdout, "Hello, world!\n");
     Ok(())
 }
@@ -1104,7 +1106,7 @@ fn increase_stack_size() -> Result<()> {
 
 mod test_programs {
     use super::{get_wasmtime_command, run_wasmtime};
-    use anyhow::{Context, Result, bail};
+    use anyhow::{Context, Result, bail, format_err};
     use http_body_util::BodyExt;
     use hyper::header::HeaderValue;
     use std::io::{self, BufRead, BufReader, Read, Write};
@@ -1121,7 +1123,7 @@ mod test_programs {
             use self::$name as _;
         };
     }
-    foreach_p2_cli!(assert_test_exists);
+    foreach_cli!(assert_test_exists);
 
     #[test]
     fn p2_cli_hello_stdout() -> Result<()> {
@@ -2149,7 +2151,7 @@ start a print 1234
     }
 
     #[test]
-    fn p2_cli_p3_hello_stdout() -> Result<()> {
+    fn p3_cli_hello_stdout() -> Result<()> {
         let output = run_wasmtime(&[
             "run",
             "-Wcomponent-model-async",
@@ -2165,24 +2167,146 @@ start a print 1234
         Ok(())
     }
 
-    mod invoke {
-        use super::*;
+    #[test]
+    fn p2_cli_hello_stdout_invoke() -> Result<()> {
+        println!("{P2_CLI_HELLO_STDOUT_COMPONENT}");
+        let output = run_wasmtime(&[
+            "run",
+            "-Wcomponent-model",
+            "--invoke",
+            "run()",
+            P2_CLI_HELLO_STDOUT_COMPONENT,
+        ])?;
+        // First this component prints "hello, world", then the invoke
+        // result is printed as "ok".
+        assert_eq!(output, "hello, world\nok\n");
+        Ok(())
+    }
+
+    #[test]
+    fn p2_random_limits() -> Result<()> {
+        println!("{P2_RANDOM_COMPONENT}");
+
+        // By default, p2_random.rs fits within the random limits - always
+        // asks for 256 bytes.
+        let output = run_wasmtime(&["run", P2_RANDOM_COMPONENT])?;
+        assert_eq!(output, "");
+
+        // Lowering limit to 255 bytes should produce a trap in the first
+        // call, which goes by way of the wasip1 adapter:
+        let output = run_wasmtime(&["run", "-Smax-random-size=255", P2_RANDOM_COMPONENT])
+            .err()
+            .ok_or_else(|| format_err!("execution with max-random-size=255 should trap"))?;
+        let output = format!("{output:?}");
+        assert!(
+            output.contains("lib_generated::random_get"),
+            "expected error stack frames to contain 'wasip1::lib_generated::random_get'. Got:\n{output}"
+        );
+        assert!(
+            output.contains("requested len 256 exceeds limit 255"),
+            "expected error stack frames to contain 'requested len 256 exceeds limit 255'. Got:\n{output}"
+        );
+
+        // Lowering limit to 255 bytes and setting environment variable for
+        // the first call to only request 255 bytes should be OK, and then the
+        // program produce a trap in the second call, which calls the wasip2
+        // random import directly:
+        let output = run_wasmtime(&[
+            "run",
+            "-Smax-random-size=255",
+            "--env",
+            "TEST_P1_RANDOM_LEN=255",
+            P2_RANDOM_COMPONENT,
+        ])
+        .err()
+        .ok_or_else(|| format_err!("execution with max-random-size=255 should trap"))?;
+        let output = format!("{output:?}");
+        assert!(
+            output.contains("random::random::get_random_bytes"),
+            "expected error stack frames to contain 'wasi::random::random::get_random_bytes'. Got:\n{output}"
+        );
+        assert!(
+            output.contains("requested len 256 exceeds limit 255"),
+            "expected error stack frames to contain 'requested len 256 exceeds limit 255'. Got:\n{output}"
+        );
+
+        // Lowering limit to 255 bytes and setting environment variable for
+        // the first and second calls to be at the limit should produce a trap
+        // in the third call, which calls the wasip2 insecure random import:
+        let output = run_wasmtime(&[
+            "run",
+            "-Smax-random-size=255",
+            "--env",
+            "TEST_P1_RANDOM_LEN=255",
+            "--env",
+            "TEST_P2_RANDOM_LEN=255",
+            P2_RANDOM_COMPONENT,
+        ])
+        .err()
+        .ok_or_else(|| format_err!("execution with max-random-size=255 should trap"))?;
+        let output = format!("{output:?}");
+        assert!(
+            output.contains("random::insecure::get_insecure_random_bytes"),
+            "expected error stack frames to contain 'wasi::random::insecure::get_insecure_random_bytes'. Got:\n{output}"
+        );
+        assert!(
+            output.contains("requested len 256 exceeds limit 255"),
+            "expected error stack frames to contain 'requested len 256 exceeds limit 255'. Got:\n{output}"
+        );
+
+        // Lowering limit to 255 bytes and setting environment variable for
+        // the first and second calls to be under the limit should trap in the
+        // third call, which calls the wasip2 insecure random import:
+        let output = run_wasmtime(&[
+            "run",
+            "-Smax-random-size=255",
+            "--env",
+            "TEST_P1_RANDOM_LEN=255",
+            "--env",
+            "TEST_P2_RANDOM_LEN=255",
+            "--env",
+            "TEST_P2_INSECURE_RANDOM_LEN=255",
+            P2_RANDOM_COMPONENT,
+        ])
+        .context("setting all calls to be equal to the limit should pass")?;
+        assert_eq!(output, "");
+
+        Ok(())
+    }
 
-        #[test]
-        fn p2_cli_hello_stdout() -> Result<()> {
-            println!("{P2_CLI_HELLO_STDOUT_COMPONENT}");
-            let output = run_wasmtime(&[
+    #[test]
+    fn p2_cli_http_headers() -> Result<()> {
+        for test in ["append", "append-empty", "append-same", "append-same-empty"] {
+            let err = run_wasmtime(&[
                 "run",
-                "-Wcomponent-model",
-                "--invoke",
-                "run()",
-                P2_CLI_HELLO_STDOUT_COMPONENT,
-            ])?;
-            // First this component prints "hello, world", then the invoke
-            // result is printed as "ok".
-            assert_eq!(output, "hello, world\nok\n");
-            Ok(())
+                "-Shttp",
+                "-Smax-http-fields-size=1048576",
+                P2_CLI_HTTP_HEADERS_COMPONENT,
+                test,
+            ])
+            .unwrap_err();
+            assert!(
+                err.to_string()
+                    .contains("Field size limit 1048576 exceeded")
+                    || err.to_string().contains("max size reached"),
+                "bad error message: {err:?}"
+            );
         }
+
+        // With an extremely large limit Wasmtime still shouldn't panic.
+        let err = run_wasmtime(&[
+            "run",
+            "-Shttp",
+            &format!("-Smax-http-fields-size={}", 1 << 30),
+            P2_CLI_HTTP_HEADERS_COMPONENT,
+            "append",
+        ])
+        .unwrap_err();
+        assert!(
+            err.to_string().contains("max size reached"),
+            "bad error message: {err:?}"
+        );
+        Ok(())
     }
 
     #[test]
@@ -2245,6 +2369,18 @@ start a print 1234
         )
     }
 
+    #[test]
+    fn p2_cli_max_resources() -> Result<()> {
+        let err = run_wasmtime(&["run", "-Smax-resources=50", P2_CLI_MAX_RESOURCES_COMPONENT])
+            .unwrap_err();
+        assert!(
+            err.to_string().contains("resource table has no free keys"),
+            "bad error message: {err}"
+        );
+        run_wasmtime(&["run", "-Smax-resources=200", P2_CLI_MAX_RESOURCES_COMPONENT])?;
+        Ok(())
+    }
+
     #[tokio::test]
     #[cfg_attr(not(feature = "component-model-async"), ignore)]
     async fn p3_cli_serve_hello_world() -> Result<()> {
@@ -2485,6 +2621,98 @@ start a print 1234
         assert!(stderr.contains("guest timed out"), "bad stderr: {stderr}");
         Ok(())
     }
+
+    #[test]
+    fn p2_cli_many_resources() -> Result<()> {
+        let err = run_wasmtime(&["run", P2_CLI_MANY_RESOURCES_COMPONENT]).unwrap_err();
+        assert!(
+            err.to_string().contains("resource table has no free keys"),
+            "bad error message: {err}"
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn p1_cli_hostcall_fuel() -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        run_wasmtime(&[
+            "run",
+            &format!("--dir={}::.", dir.path().to_str().unwrap()),
+            "-Shostcall-fuel=1000",
+            P1_CLI_HOSTCALL_FUEL,
+        ])?;
+        Ok(())
+    }
+
+    #[test]
+    fn p2_cli_hostcall_fuel() -> Result<()> {
+        enum Exit {
+            Ok,
+            NoFuel,
+            TooManyZeroes,
+            BufferTooLarge,
+        }
+
+        let dir = tempfile::tempdir()?;
+        let file = dir.path().join("1mb");
+        std::fs::write(&file, vec![0; 1024 * 1024])?;
+        for (arg, exit) in [
+            ("poll", Exit::NoFuel),
+            ("read", Exit::Ok),
+            ("write", Exit::NoFuel),
+            ("mkdir", Exit::NoFuel),
+            ("write-stream", Exit::NoFuel),
+            ("write-stream-blocking", Exit::NoFuel),
+            ("resolve", Exit::NoFuel),
+            ("udp-send-many", Exit::NoFuel),
+            ("udp-send-big", Exit::NoFuel),
+            ("write-zeroes", Exit::TooManyZeroes),
+            ("write-stream-buffer-too-large", Exit::BufferTooLarge),
+            ("write-zeroes-buffer-too-large", Exit::BufferTooLarge),
+            ("read-file-big", Exit::Ok),
+            ("read-tcp-big", Exit::Ok),
+        ] {
+            println!("test: {arg}");
+            let result = run_wasmtime(&[
+                "run",
+                "-Shostcall-fuel=5000",
+                "-Sinherit-network",
+                &format!("--dir={}::.", dir.path().to_str().unwrap()),
+                P2_CLI_HOSTCALL_FUEL_COMPONENT,
+                arg,
+            ]);
+
+            match exit {
+                Exit::Ok => {
+                    result.unwrap();
+                }
+                Exit::NoFuel => {
+                    let err = result.unwrap_err();
+                    assert!(
+                        err.to_string()
+                            .contains("fuel allocated for hostcalls has been exhausted"),
+                        "bad error message: {err}"
+                    );
+                }
+                Exit::TooManyZeroes => {
+                    let err = result.unwrap_err();
+                    assert!(
+                        err.to_string()
+                            .contains("cannot write more zeroes than `check_write` allows"),
+                        "bad error message: {err}"
+                    );
+                }
+                Exit::BufferTooLarge => {
+                    let err = result.unwrap_err();
+                    assert!(
+                        err.to_string().contains("Buffer too large"),
+                        "bad error message: {err}"
+                    );
+                }
+            }
+        }
+        Ok(())
+    }
 }
 
 #[test]

@@ -3316,3 +3316,51 @@ fn run_export_with_internal_adapter() -> Result<()> {
     assert_eq!(run.call(&mut store, ())?, (5,));
     Ok(())
 }
+
+#[tokio::test]
+async fn drop_call_async_future() -> Result<()> {
+    let component = r#"
+(component
+  (import "foo" (func $f))
+  (core module $m
+    (func $f (import "" "foo"))
+    (func (export "foo") call $f)
+  )
+  (core func $f (canon lower (func $f)))
+  (core instance $m (instantiate $m (with "" (instance
+     (export "foo" (func $f))
+  ))))
+  (func (export "foo") (canon lift (core func $m "foo")))
+)
+"#;
+
+    let mut config = Config::new();
+    config.async_support(true);
+    let engine = &Engine::new(&config)?;
+    let component = Component::new(&engine, component)?;
+    let mut store = Store::new(&engine, ());
+    let mut linker = Linker::new(&engine);
+    linker.root().func_wrap_async("foo", |_, _: ()| {
+        Box::new(async {
+            tokio::task::yield_now().await;
+            Ok(())
+        })
+    })?;
+    let instance = linker.instantiate_async(&mut store, &component).await?;
+    let foo = instance.get_typed_func::<(), ()>(&mut store, "foo")?;
+    // Here we'll use `call_async` a few times but only poll each returned
+    // future once.  This will put the instance in a weird state but shouldn't
+    // cause a panic.
+    for _ in 0..5 {
+        let mut future = std::pin::pin!(async {
+            foo.call_async(&mut store, ()).await?;
+            foo.post_return_async(&mut store).await
+        });
+        if let std::task::Poll::Ready(result) =
+            std::future::poll_fn(|cx| std::task::Poll::Ready(future.as_mut().poll(cx))).await
+        {
+            _ = result;
+        }
+    }
+    Ok(())
+}

@@ -1203,3 +1203,47 @@ fn use_types_across_component_boundaries() -> Result<()> {
 
     Ok(())
 }
+
+#[test]
+fn hostcall_fuel_limits_val() -> Result<()> {
+    let engine = super::engine();
+    let component = Component::new(
+        &engine,
+        r#"(component
+            (import "hi" (func $hi (param "x" (list u8))))
+            (core module $libc
+                (memory (export "memory") 10)
+            )
+            (core module $m
+                (import "libc" "memory" (memory 1))
+                (import "" "hi" (func $hi (param i32 i32)))
+
+                (func (export "run")
+                    i32.const 0
+                    memory.size
+                    i32.const 65536
+                    i32.mul
+                    call $hi)
+            )
+            (core instance $libc (instantiate $libc))
+            (core func $hi (canon lower (func $hi) (memory $libc "memory")))
+            (core instance $i (instantiate $m
+                (with "libc" (instance $libc))
+                (with "" (instance (export "hi" (func $hi))))
+            ))
+            (func (export "run") (canon lift (core func $i "run")))
+        )"#,
+    )?;
+    let mut store = Store::new(&engine, 0);
+    let mut linker = Linker::new(&engine);
+    linker.root().func_new("hi", |_, _, _, _| Ok(()))?;
+    let instance = linker.instantiate(&mut store, &component)?;
+    let run = instance.get_func(&mut store, "run").unwrap();
+    run.call(&mut store, &[], &mut [])?;
+    run.post_return(&mut store)?;
+
+    store.set_hostcall_fuel(100);
+    assert!(run.call(&mut store, &[], &mut []).is_err());
+
+    Ok(())
+}

@@ -492,6 +492,18 @@ wasmtime_option_group! {
         pub keyvalue_in_memory_data: Vec<KeyValuePair>,
         /// Enable support for WASIp3 APIs.
         pub p3: Option<bool>,
+        /// Maximum resources the guest is allowed to create simultaneously.
+        pub max_resources: Option<usize>,
+        /// Fuel to use for all hostcalls to limit guest<->host data transfer.
+        pub hostcall_fuel: Option<usize>,
+        /// Maximum value, in bytes, for a wasi-random 0.2
+        /// `get{,-insecure}-random-bytes` `len` parameter. Calls with a value
+        /// exceeding this limit will trap.
+        pub max_random_size: Option<u64>,
+        /// Maximum value, in bytes, for the contents of a wasi-http 0.2
+        /// `fields` resource (aka `headers` and `trailers`). `fields` methods
+        /// which cause the contents to exceed this size limit will trap.
+        pub max_http_fields_size: Option<usize>,
     }
 
     enum Wasi {

@@ -73,21 +73,26 @@ impl Artifacts {
             // Bucket, based on the name of the test, into a "kind" which
             // generates a `foreach_*` macro below.
             let kind = match test.name.as_str() {
+                s if s.starts_with("p1_cli_")
+                    || s.starts_with("p2_cli_")
+                    || s.starts_with("p3_cli_") =>
+                {
+                    "cli"
+                }
                 s if s.starts_with("p1_") => "p1",
                 s if s.starts_with("p2_http_") => "p2_http",
-                s if s.starts_with("p2_cli_") => "p2_cli",
                 s if s.starts_with("p2_api_") => "p2_api",
                 s if s.starts_with("p2_") => "p2",
+                s if s.starts_with("p3_http_") => "p3_http",
+                s if s.starts_with("p3_api_") => "p3_api",
+                s if s.starts_with("p3_") => "p3",
                 s if s.starts_with("nn_") => "nn",
                 s if s.starts_with("piped_") => "piped",
                 s if s.starts_with("dwarf_") => "dwarf",
                 s if s.starts_with("config_") => "config",
                 s if s.starts_with("keyvalue_") => "keyvalue",
                 s if s.starts_with("tls_") => "tls",
                 s if s.starts_with("async_") => "async",
-                s if s.starts_with("p3_http_") => "p3_http",
-                s if s.starts_with("p3_api_") => "p3_api",
-                s if s.starts_with("p3_") => "p3",
                 s if s.starts_with("fuzz_") => "fuzz",
                 // If you're reading this because you hit this panic, either add
                 // it to a test suite above or add a new "suite". The purpose of
@@ -272,7 +277,7 @@ impl Artifacts {
                 // Prevent stray files for now that we don't understand.
                 Some(_) => panic!("unknown file extension on {path:?}"),
 
-                None => unreachable!(),
+                None => unreachable!("no extension in path {path:?}"),
             }
         }
     }

@@ -0,0 +1,118 @@
+use std::ptr;
+
+fn main() {
+    big_poll();
+    big_string();
+    big_iovecs();
+}
+
+fn big_string() {
+    let mut s = String::new();
+    for _ in 0..10_000 {
+        s.push_str("hello world");
+    }
+    let dir_fd = test_programs::preview1::open_scratch_directory(".").unwrap();
+    assert_eq!(
+        unsafe { wasip1::path_create_directory(dir_fd, &s) },
+        Err(wasip1::ERRNO_NOMEM)
+    );
+}
+
+fn big_iovecs() {
+    let mut iovs = Vec::new();
+    let mut ciovs = Vec::new();
+    for _ in 0..10_000 {
+        iovs.push(wasip1::Iovec {
+            buf: ptr::null_mut(),
+            buf_len: 0,
+        });
+        ciovs.push(wasip1::Ciovec {
+            buf: ptr::null(),
+            buf_len: 0,
+        });
+    }
+    let dir_fd = test_programs::preview1::open_scratch_directory(".").unwrap();
+    let fd = unsafe {
+        wasip1::path_open(
+            dir_fd,
+            0,
+            "hi",
+            wasip1::OFLAGS_CREAT,
+            wasip1::RIGHTS_FD_WRITE | wasip1::RIGHTS_FD_READ,
+            0,
+            0,
+        )
+        .unwrap()
+    };
+
+    unsafe {
+        assert_eq!(wasip1::fd_write(fd, &ciovs), Err(wasip1::ERRNO_NOMEM));
+        assert_eq!(wasip1::fd_read(fd, &iovs), Err(wasip1::ERRNO_NOMEM));
+        assert_eq!(wasip1::fd_pwrite(fd, &ciovs, 0), Err(wasip1::ERRNO_NOMEM));
+        assert_eq!(wasip1::fd_pread(fd, &iovs, 0), Err(wasip1::ERRNO_NOMEM));
+    }
+
+    ciovs.truncate(1);
+    iovs.truncate(1);
+    iovs.push(wasip1::Iovec {
+        buf: ptr::null_mut(),
+        buf_len: 10_000,
+    });
+    ciovs.push(wasip1::Ciovec {
+        buf: ptr::null(),
+        buf_len: 10_000,
+    });
+    unsafe {
+        assert_eq!(wasip1::fd_write(fd, &ciovs), Err(wasip1::ERRNO_NOMEM));
+        assert_eq!(wasip1::fd_read(fd, &iovs), Err(wasip1::ERRNO_NOMEM));
+        assert_eq!(wasip1::fd_pwrite(fd, &ciovs, 0), Err(wasip1::ERRNO_NOMEM));
+        assert_eq!(wasip1::fd_pread(fd, &iovs, 0), Err(wasip1::ERRNO_NOMEM));
+    }
+}
+
+fn big_poll() {
+    let mut huge_poll = Vec::new();
+    let mut huge_events = Vec::new();
+    for _ in 0..10_000 {
+        huge_poll.push(subscribe_timeout(0));
+        huge_events.push(empty_event());
+    }
+    let err = unsafe {
+        wasip1::poll_oneoff(
+            huge_poll.as_ptr(),
+            huge_events.as_mut_ptr(),
+            huge_poll.len(),
+        )
+        .unwrap_err()
+    };
+    assert_eq!(err, wasip1::ERRNO_NOMEM);
+
+    fn subscribe_timeout(timeout: u64) -> wasip1::Subscription {
+        wasip1::Subscription {
+            userdata: 0,
+            u: wasip1::SubscriptionU {
+                tag: wasip1::EVENTTYPE_CLOCK.raw(),
+                u: wasip1::SubscriptionUU {
+                    clock: wasip1::SubscriptionClock {
+                        id: wasip1::CLOCKID_MONOTONIC,
+                        timeout,
+                        precision: 0,
+                        flags: 0,
+                    },
+                },
+            },
+        }
+    }
+
+    fn empty_event() -> wasip1::Event {
+        wasip1::Event {
+            error: wasip1::ERRNO_SUCCESS,
+            fd_readwrite: wasip1::EventFdReadwrite {
+                nbytes: 0,
+                flags: 0,
+            },
+            type_: wasip1::EVENTTYPE_CLOCK,
+            userdata: 0,
+        }
+    }
+}

@@ -0,0 +1,98 @@
+use std::process;
+use test_programs::preview1::open_scratch_directory;
+
+fn main() {
+    let arg = std::env::args().nth(1).unwrap();
+    let dir_fd = match open_scratch_directory(&arg) {
+        Ok(dir_fd) => dir_fd,
+        Err(err) => {
+            eprintln!("{err}");
+            process::exit(1)
+        }
+    };
+
+    // Wait for just one timeout (maybe hitting a fast path)
+    let subs = [subscribe_timeout(0)];
+    for _ in 0..1000 {
+        let mut events = [empty_event()];
+        unsafe {
+            wasip1::poll_oneoff(subs.as_ptr(), events.as_mut_ptr(), 1).unwrap();
+        }
+    }
+
+    // Wait for two timeouts
+    let subs = [subscribe_timeout(0), subscribe_timeout(0)];
+    for _ in 0..1000 {
+        let mut events = [empty_event(), empty_event()];
+        unsafe {
+            wasip1::poll_oneoff(subs.as_ptr(), events.as_mut_ptr(), 2).unwrap();
+        }
+    }
+
+    let file_fd = unsafe {
+        wasip1::path_open(
+            dir_fd,
+            0,
+            "hello.txt",
+            wasip1::OFLAGS_CREAT,
+            wasip1::RIGHTS_FD_WRITE | wasip1::RIGHTS_FD_READ,
+            0,
+            0,
+        )
+        .expect("creating a file for writing")
+    };
+
+    // Wait for a timeout fd operations
+    let subs = [
+        subscribe_timeout(0),
+        subscribe_fd(wasip1::EVENTTYPE_FD_READ, file_fd),
+        subscribe_fd(wasip1::EVENTTYPE_FD_WRITE, file_fd),
+    ];
+    for _ in 0..1000 {
+        let mut events = [empty_event(), empty_event(), empty_event()];
+        unsafe {
+            wasip1::poll_oneoff(subs.as_ptr(), events.as_mut_ptr(), 3).unwrap();
+        }
+    }
+}
+
+fn subscribe_timeout(timeout: u64) -> wasip1::Subscription {
+    wasip1::Subscription {
+        userdata: 0,
+        u: wasip1::SubscriptionU {
+            tag: wasip1::EVENTTYPE_CLOCK.raw(),
+            u: wasip1::SubscriptionUU {
+                clock: wasip1::SubscriptionClock {
+                    id: wasip1::CLOCKID_MONOTONIC,
+                    timeout,
+                    precision: 0,
+                    flags: 0,
+                },
+            },
+        },
+    }
+}
+
+fn subscribe_fd(ty: wasip1::Eventtype, file_descriptor: wasip1::Fd) -> wasip1::Subscription {
+    wasip1::Subscription {
+        userdata: 0,
+        u: wasip1::SubscriptionU {
+            tag: ty.raw(),
+            u: wasip1::SubscriptionUU {
+                fd_read: wasip1::SubscriptionFdReadwrite { file_descriptor },
+            },
+        },
+    }
+}
+
+fn empty_event() -> wasip1::Event {
+    wasip1::Event {
+        error: wasip1::ERRNO_SUCCESS,
+        fd_readwrite: wasip1::EventFdReadwrite {
+            nbytes: 0,
+            flags: 0,
+        },
+        type_: wasip1::EVENTTYPE_CLOCK,
+        userdata: 0,
+    }
+}

@@ -1,3 +1,4 @@
+use anyhow::{Context, Result};
 use test_programs::wasi::http::types::{
     Headers, IncomingRequest, Method, OutgoingBody, OutgoingResponse, ResponseOutparam,
 };
@@ -29,6 +30,16 @@ impl test_programs::proxy::exports::wasi::http::incoming_handler::Guest for T {
 
                 return;
             }
+            (Method::Get, Some(p)) if p.starts_with("/modify_fields/") => {
+                let r = modify_fields_handler(request);
+                response_for(r, outparam);
+                return;
+            }
+            (Method::Get, Some(p)) if p.starts_with("/new_fields/") => {
+                let r = new_fields_handler(request);
+                response_for(r, outparam);
+                return;
+            }
 
             _ => {}
         }
@@ -69,10 +80,64 @@ impl test_programs::proxy::exports::wasi::http::incoming_handler::Guest for T {
     }
 }
 
+fn response_for(r: Result<()>, outparam: ResponseOutparam) {
+    let resp = OutgoingResponse::new(Headers::new());
+    resp.set_status_code(if r.is_ok() { 200 } else { 500 })
+        .unwrap();
+    let body = resp.body().expect("outgoing response");
+    ResponseOutparam::set(outparam, Ok(resp));
+    let _ = body.write().and_then(|out| {
+        let _ = out.blocking_write_and_flush(format!("{r:?}").as_bytes());
+        drop(out);
+        Ok(())
+    });
+    let _ = OutgoingBody::finish(body, None);
+}
+
 // Technically this should not be here for a proxy, but given the current
 // framework for tests it's required since this file is built as a `bin`
 fn main() {}
 
 fn test_filesystem() {
     assert!(std::fs::File::open(".").is_err());
 }
+
+fn add_bytes_to_headers(headers: Headers, size: usize) {
+    if size == 0 {
+        return;
+    } else if size < 10 {
+        headers.append("k", &b"abcdefghi"[0..size - 1]).unwrap()
+    } else {
+        for chunk in 0..(size / 10) {
+            let k = format!("g{chunk:04}");
+            let mut v = format!("h{chunk:04}");
+            if chunk == 0 {
+                for _ in 0..(size % 10) {
+                    v.push('#');
+                }
+            }
+            headers.append(k.as_str(), v.as_bytes()).unwrap()
+        }
+    }
+}
+
+fn modify_fields_handler(request: IncomingRequest) -> Result<()> {
+    let path = request.path_with_query().unwrap();
+    let rest = path.trim_start_matches("/modify_fields/");
+    let added_field_bytes: usize = rest
+        .parse()
+        .context("expect remainder of url to parse as number")?;
+    add_bytes_to_headers(request.headers().clone(), added_field_bytes);
+
+    Ok(())
+}
+fn new_fields_handler(request: IncomingRequest) -> Result<()> {
+    let path = request.path_with_query().unwrap();
+    let rest = path.trim_start_matches("/new_fields/");
+    let added_field_bytes: usize = rest
+        .parse()
+        .context("expect remainder of url to parse as number")?;
+    add_bytes_to_headers(Headers::new(), added_field_bytes);
+
+    Ok(())
+}

@@ -0,0 +1,282 @@
+use wasip2::filesystem::types::*;
+use wasip2::sockets::network::*;
+
+fn main() {
+    match std::env::args().nth(1).as_deref() {
+        Some("poll") => poll(),
+        Some("read") => read(),
+        Some("write") => write(),
+        Some("write-stream") => write_stream(),
+        Some("write-stream-blocking") => write_stream_blocking(),
+        Some("mkdir") => mkdir(),
+        Some("resolve") => resolve(),
+        Some("udp-send-many") => udp_send_many(),
+        Some("udp-send-big") => udp_send_big(),
+        Some("write-zeroes") => write_zeroes(),
+        Some("write-stream-buffer-too-large") => write_stream_buffer_too_large(),
+        Some("write-zeroes-buffer-too-large") => write_zeroes_buffer_too_large(),
+        Some("read-file-big") => read_file_big(),
+        Some("read-tcp-big") => read_tcp_big(),
+        other => panic!("unknown arg {other:?}"),
+    }
+}
+
+fn poll() {
+    let mut events = Vec::new();
+    let sub = wasip2::clocks::monotonic_clock::subscribe_duration(0);
+    for _ in 0..5000 {
+        events.push(&sub);
+    }
+
+    wasip2::io::poll::poll(&events);
+    unreachable!()
+}
+
+fn preopen() -> Descriptor {
+    let mut dirs = wasip2::filesystem::preopens::get_directories();
+    assert_eq!(dirs.len(), 1);
+    dirs.pop().unwrap().0
+}
+
+fn read() {
+    let f = preopen()
+        .open_at(
+            PathFlags::empty(),
+            "1mb",
+            OpenFlags::empty(),
+            DescriptorFlags::empty(),
+        )
+        .unwrap();
+
+    // 0-length is ok
+    f.read(0, 0).unwrap();
+
+    // This isn't transferring data from the guest to the host, so this is ok.
+    f.read(1 << 20, 0).unwrap();
+
+    // Host shouldn't allocate based on what the guest asks for...
+    f.read(u64::MAX, 0).unwrap();
+
+    f.read_via_stream(0)
+        .unwrap()
+        .blocking_read(1 << 20)
+        .unwrap();
+}
+
+fn write() {
+    let f = preopen()
+        .open_at(
+            PathFlags::empty(),
+            "hi",
+            OpenFlags::CREATE,
+            DescriptorFlags::empty(),
+        )
+        .unwrap();
+    f.write(&[0; 5001], 0).unwrap();
+    unreachable!()
+}
+
+fn write_stream() {
+    preopen()
+        .open_at(
+            PathFlags::empty(),
+            "hi",
+            OpenFlags::CREATE,
+            DescriptorFlags::empty(),
+        )
+        .unwrap()
+        .write_via_stream(0)
+        .unwrap()
+        .write(&[0; 5001])
+        .unwrap();
+    unreachable!()
+}
+
+fn write_stream_blocking() {
+    preopen()
+        .open_at(
+            PathFlags::empty(),
+            "hi",
+            OpenFlags::CREATE,
+            DescriptorFlags::empty(),
+        )
+        .unwrap()
+        .write_via_stream(0)
+        .unwrap()
+        .blocking_write_and_flush(&[0; 5001])
+        .unwrap();
+    unreachable!()
+}
+
+fn mkdir() {
+    let mut name = String::new();
+    for _ in 0..5001 {
+        name.push_str("a");
+    }
+    preopen().create_directory_at(&name).unwrap();
+    unreachable!()
+}
+
+fn resolve() {
+    let network = wasip2::sockets::instance_network::instance_network();
+    let mut name = String::new();
+    for _ in 0..5001 {
+        name.push_str("a");
+    }
+    wasip2::sockets::ip_name_lookup::resolve_addresses(&network, &name).unwrap();
+    unreachable!();
+}
+
+fn udp_socket() -> wasip2::sockets::udp::UdpSocket {
+    let socket =
+        wasip2::sockets::udp_create_socket::create_udp_socket(IpAddressFamily::Ipv4).unwrap();
+    let network = wasip2::sockets::instance_network::instance_network();
+
+    socket
+        .start_bind(
+            &network,
+            IpSocketAddress::Ipv4(Ipv4SocketAddress {
+                address: (127, 0, 0, 1),
+                port: 0,
+            }),
+        )
+        .unwrap();
+    socket.finish_bind().unwrap();
+    socket
+}
+
+fn udp_send_many() {
+    let socket = udp_socket();
+    let (_incoming, outgoing) = socket.stream(None).unwrap();
+    let mut dgrams = Vec::new();
+
+    for _ in 0..5000 {
+        dgrams.push(wasip2::sockets::udp::OutgoingDatagram {
+            data: Vec::new(),
+            remote_address: None,
+        });
+    }
+
+    outgoing.send(&dgrams).unwrap();
+    unreachable!()
+}
+
+fn udp_send_big() {
+    let socket = udp_socket();
+    let (_incoming, outgoing) = socket.stream(None).unwrap();
+    let mut dgrams = Vec::new();
+
+    for _ in 0..2 {
+        dgrams.push(wasip2::sockets::udp::OutgoingDatagram {
+            data: vec![0; 2500],
+            remote_address: None,
+        });
+    }
+
+    outgoing.send(&dgrams).unwrap();
+    unreachable!()
+}
+
+fn write_zeroes() {
+    preopen()
+        .open_at(
+            PathFlags::empty(),
+            "hi",
+            OpenFlags::CREATE,
+            DescriptorFlags::empty(),
+        )
+        .unwrap()
+        .write_via_stream(0)
+        .unwrap()
+        .write_zeroes(u64::MAX)
+        .unwrap();
+    unreachable!()
+}
+
+fn write_stream_buffer_too_large() {
+    preopen()
+        .open_at(
+            PathFlags::empty(),
+            "hi",
+            OpenFlags::CREATE,
+            DescriptorFlags::empty(),
+        )
+        .unwrap()
+        .write_via_stream(0)
+        .unwrap()
+        .blocking_write_and_flush(&[0; 5000])
+        .unwrap();
+    unreachable!()
+}
+
+fn write_zeroes_buffer_too_large() {
+    preopen()
+        .open_at(
+            PathFlags::empty(),
+            "hi",
+            OpenFlags::CREATE,
+            DescriptorFlags::empty(),
+        )
+        .unwrap()
+        .write_via_stream(0)
+        .unwrap()
+        .blocking_write_zeroes_and_flush(5000)
+        .unwrap();
+    unreachable!()
+}
+
+fn read_file_big() {
+    preopen()
+        .open_at(
+            PathFlags::empty(),
+            "1mb",
+            OpenFlags::empty(),
+            DescriptorFlags::empty(),
+        )
+        .unwrap()
+        .read_via_stream(0)
+        .unwrap()
+        .blocking_read(u64::MAX)
+        .unwrap();
+}
+
+fn read_tcp_big() {
+    let server =
+        wasip2::sockets::tcp_create_socket::create_tcp_socket(IpAddressFamily::Ipv4).unwrap();
+    let client =
+        wasip2::sockets::tcp_create_socket::create_tcp_socket(IpAddressFamily::Ipv4).unwrap();
+
+    server
+        .start_bind(
+            &wasip2::sockets::instance_network::instance_network(),
+            IpSocketAddress::Ipv4(Ipv4SocketAddress {
+                address: (127, 0, 0, 1),
+                port: 0,
+            }),
+        )
+        .unwrap();
+    server.finish_bind().unwrap();
+    server.start_listen().unwrap();
+    server.finish_listen().unwrap();
+
+    client
+        .start_connect(
+            &wasip2::sockets::instance_network::instance_network(),
+            server.local_address().unwrap(),
+        )
+        .unwrap();
+    client.subscribe().block();
+    let (input, _output) = client.finish_connect().unwrap();
+
+    {
+        server.subscribe().block();
+        let (socket, input, output) = server.accept().unwrap();
+        drop((input, output));
+        drop(socket);
+    }
+
+    match input.blocking_read(u64::MAX) {
+        Err(wasip2::io::streams::StreamError::Closed) => {}
+        other => panic!("unexpected result: {other:?}"),
+    }
+}

@@ -0,0 +1,33 @@
+fn main() {
+    let fields = wasip2::http::types::Fields::new();
+
+    match std::env::args().nth(1).as_deref() {
+        Some("append") => {
+            for i in 0.. {
+                if fields.append(&format!("a{i}"), b"a").is_err() {
+                    break;
+                }
+            }
+        }
+        Some("append-empty") => {
+            for i in 0.. {
+                if fields.append(&format!("a{i}"), b"").is_err() {
+                    break;
+                }
+            }
+        }
+        Some("append-same") => loop {
+            if fields.append("a", b"b").is_err() {
+                break;
+            }
+        },
+        Some("append-same-empty") => loop {
+            if fields.append("a", b"").is_err() {
+                break;
+            }
+        },
+        other => panic!("unknown test {other:?}"),
+    }
+
+    unreachable!();
+}

@@ -0,0 +1,7 @@
+use test_programs::wasi::clocks::monotonic_clock::subscribe_duration;
+
+fn main() {
+    loop {
+        std::mem::forget(subscribe_duration(1_000_000));
+    }
+}

@@ -0,0 +1,6 @@
+fn main() {
+    let mut buf = Vec::new();
+    for _ in 0..100 {
+        buf.push(wasip2::clocks::monotonic_clock::subscribe_duration(0));
+    }
+}

@@ -1,16 +1,23 @@
+use std::mem::MaybeUninit;
 use test_programs::wasi::random;
 
 fn main() {
-    let mut bytes = [0_u8; 256];
+    let p1_random_size: usize = std::env::var("TEST_P1_RANDOM_LEN")
+        .map(|v| v.parse().expect("TEST_P1_RANDOM_LEN should be a usize"))
+        .unwrap_or(256);
+    let mut bytes: Box<[MaybeUninit<u8>]> = Box::new_uninit_slice(p1_random_size);
     unsafe {
-        wasip1::random_get(bytes.as_mut_ptr(), bytes.len()).unwrap();
+        wasip1::random_get(bytes.as_mut_ptr() as *mut u8, bytes.len()).unwrap();
     }
 
-    assert!(bytes.iter().any(|x| *x != 0));
+    assert!(bytes.iter().any(|x| unsafe { x.assume_init() } != 0));
 
+    let p2_random_size: u64 = std::env::var("TEST_P2_RANDOM_LEN")
+        .map(|v| v.parse().expect("TEST_P2_RANDOM_LEN should be a u64"))
+        .unwrap_or(256);
     // Acquired random bytes should be of the expected length.
-    let array = random::random::get_random_bytes(100);
-    assert_eq!(array.len(), 100);
+    let array = random::random::get_random_bytes(p2_random_size);
+    assert_eq!(array.len(), p2_random_size as usize);
 
     // It shouldn't take 100+ tries to get a nonzero random integer.
     for i in 0.. {
@@ -27,9 +34,15 @@ fn main() {
     assert_eq!(a1, a2);
     assert_eq!(b1, b2);
 
+    let p2_insecure_random_size: u64 = std::env::var("TEST_P2_INSECURE_RANDOM_LEN")
+        .map(|v| {
+            v.parse()
+                .expect("TEST_P2_INSECURE_RANDOM_LEN should be a u64")
+        })
+        .unwrap_or(256);
     // Acquired random bytes should be of the expected length.
-    let array = random::insecure::get_insecure_random_bytes(100);
-    assert_eq!(array.len(), 100);
+    let array = random::insecure::get_insecure_random_bytes(p2_insecure_random_size);
+    assert_eq!(array.len(), p2_insecure_random_size as usize);
 
     // It shouldn't take 100+ tries to get a nonzero random integer.
     for i in 0.. {

@@ -27,7 +27,7 @@ log = { workspace = true }
 async-trait = { workspace = true }
 
 # Optional, enabled by wasmtime feature:
-wasmtime = { workspace = true, optional = true, features = ['runtime'] }
+wasmtime = { workspace = true, optional = true, features = ['runtime', 'component-model'] }
 # Optional, enabled by sync feature:
 cap-fs-ext = { workspace = true, optional = true }
 cap-time-ext = { workspace = true, optional = true }

@@ -5,8 +5,8 @@ use crate::{
         subscription::{RwEventFlags, Subscription},
     },
 };
-use std::future::Future;
-use std::pin::Pin;
+use std::future::{self, Future};
+use std::pin::{Pin, pin};
 use std::task::{Context, Poll as FPoll};
 
 struct FirstReady<'a, T>(Vec<Pin<Box<dyn Future<Output = T> + Send + 'a>>>);
@@ -83,13 +83,20 @@ pub async fn poll_oneoff<'a>(poll: &mut Poll<'a>) -> Result<(), Error> {
             Subscription::MonotonicClock { .. } => unreachable!(),
         }
     }
-    if let Some(Some(remaining_duration)) = duration {
-        match tokio::time::timeout(remaining_duration, futures).await {
+    match duration {
+        Some(Some(remaining)) => match tokio::time::timeout(remaining, futures).await {
             Ok(r) => r?,
             Err(_deadline_elapsed) => {}
+        },
+        Some(None) => {
+            let mut futures = pin!(futures);
+            future::poll_fn(|cx| match futures.as_mut().poll(cx) {
+                FPoll::Ready(e) => FPoll::Ready(e),
+                FPoll::Pending => FPoll::Ready(Ok(())),
+            })
+            .await?
         }
-    } else {
-        futures.await?;
+        None => futures.await?,
     }
 
     Ok(())

@@ -289,9 +289,7 @@ async fn p1_file_write() {
 async fn p1_path_open_lots() {
     run(P1_PATH_OPEN_LOTS, true).await.unwrap()
 }
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p1_cli_much_stdout() {}
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
+async fn p1_sleep_quickly_but_lots() {
+    run(P1_SLEEP_QUICKLY_BUT_LOTS, true).await.unwrap()
+}

@@ -281,9 +281,7 @@ fn p1_file_write() {
 fn p1_path_open_lots() {
     run(P1_PATH_OPEN_LOTS, true).unwrap()
 }
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p1_cli_much_stdout() {}
+#[test_log::test]
+fn p1_sleep_quickly_but_lots() {
+    run(P1_SLEEP_QUICKLY_BUT_LOTS, true).unwrap()
+}

@@ -24,6 +24,7 @@ pub type HyperOutgoingBody = UnsyncBoxBody<Bytes, types::ErrorCode>;
 #[derive(Debug)]
 pub struct HostIncomingBody {
     body: IncomingBodyState,
+    field_size_limit: usize,
     /// An optional worker task to keep alive while this body is being read.
     /// This ensures that if the parent of this body is dropped before the body
     /// then the backing data behind this worker is kept alive.
@@ -32,10 +33,15 @@ pub struct HostIncomingBody {
 
 impl HostIncomingBody {
     /// Create a new `HostIncomingBody` with the given `body` and a per-frame timeout
-    pub fn new(body: HyperIncomingBody, between_bytes_timeout: Duration) -> HostIncomingBody {
+    pub fn new(
+        body: HyperIncomingBody,
+        between_bytes_timeout: Duration,
+        field_size_limit: usize,
+    ) -> HostIncomingBody {
         let body = BodyWithTimeout::new(body, between_bytes_timeout);
         HostIncomingBody {
             body: IncomingBodyState::Start(body),
+            field_size_limit,
             worker: None,
         }
     }
@@ -155,7 +161,7 @@ enum StreamEnd {
 
     /// Body was completely read and trailers were read. Here are the trailers.
     /// Note that `None` means that the body finished without trailers.
-    Trailers(Option<FieldMap>),
+    Trailers(Option<http::HeaderMap>),
 }
 
 /// The concrete type behind the `wasi:io/streams.input-stream` resource returned
@@ -339,14 +345,15 @@ impl Pollable for HostFutureTrailers {
             match rx.await {
                 // Trailers were read for us and here they are, so store the
                 // result.
-                Ok(StreamEnd::Trailers(t)) => *self = Self::Done(Ok(t)),
-
+                Ok(StreamEnd::Trailers(Some(t))) => {
+                    *self = Self::Done(Ok(Some(FieldMap::new(t, body.field_size_limit))));
+                }
                 // The body wasn't fully read and was dropped before trailers
                 // were reached. It's up to us now to complete the body.
                 Ok(StreamEnd::Remaining(b)) => body.body = IncomingBodyState::Start(b),
 
                 // This means there were no trailers present.
-                Err(_) => {
+                Ok(StreamEnd::Trailers(None)) | Err(_) => {
                     *self = HostFutureTrailers::Done(Ok(None));
                 }
             }
@@ -370,8 +377,8 @@ impl Pollable for HostFutureTrailers {
                 Some(Ok(frame)) => {
                     // If this frame is a data frame ignore it as we're only
                     // interested in trailers.
-                    if let Ok(headers) = frame.into_trailers() {
-                        break Ok(Some(headers));
+                    if let Ok(header_map) = frame.into_trailers() {
+                        break Ok(Some(FieldMap::new(header_map, body.field_size_limit)));
                     }
                 }
             }
@@ -521,7 +528,7 @@ impl HostOutgoingBody {
         }
 
         let message = if let Some(ts) = trailers {
-            FinishMessage::Trailers(ts)
+            FinishMessage::Trailers(ts.into_inner())
         } else {
             FinishMessage::Finished
         };

@@ -79,7 +79,7 @@ where
 
         builder = builder.uri(uri.build().map_err(http_request_error)?);
 
-        for (k, v) in req.headers.iter() {
+        for (k, v) in req.headers.as_ref().iter() {
             builder = builder.header(k, v);
         }
 

@@ -3,11 +3,12 @@
 use crate::bindings::http::types::{self, Headers, Method, Scheme, StatusCode, Trailers};
 use crate::body::{HostFutureTrailers, HostIncomingBody, HostOutgoingBody, StreamContext};
 use crate::types::{
-    FieldMap, HostFields, HostFutureIncomingResponse, HostIncomingRequest, HostIncomingResponse,
-    HostOutgoingRequest, HostOutgoingResponse, HostResponseOutparam, remove_forbidden_headers,
+    FieldMap, FieldSizeLimitError, HostFields, HostFutureIncomingResponse, HostIncomingRequest,
+    HostIncomingResponse, HostOutgoingRequest, HostOutgoingResponse, HostResponseOutparam,
+    remove_forbidden_headers,
 };
 use crate::{HttpError, HttpResult, WasiHttpImpl, WasiHttpView, get_content_length};
-use anyhow::{Context, anyhow};
+use anyhow::{Context, anyhow, bail};
 use std::any::Any;
 use std::str::FromStr;
 use wasmtime::component::{Resource, ResourceTable, ResourceTableError};
@@ -80,10 +81,11 @@ where
     T: WasiHttpView,
 {
     fn new(&mut self) -> wasmtime::Result<Resource<HostFields>> {
+        let limit = self.ctx().field_size_limit;
         let id = self
             .table()
             .push(HostFields::Owned {
-                fields: hyper::HeaderMap::new(),
+                fields: FieldMap::empty(limit),
             })
             .context("[new_fields] pushing fields")?;
 
@@ -114,6 +116,14 @@ where
             fields.append(header, value);
         }
 
+        let size = FieldMap::content_size(&fields);
+        if size > self.ctx().field_size_limit {
+            bail!(FieldSizeLimitError {
+                size,
+                limit: self.ctx().field_size_limit,
+            });
+        }
+        let fields = FieldMap::new(fields, self.ctx().field_size_limit);
         let id = self
             .table()
             .push(HostFields::Owned { fields })
@@ -141,11 +151,12 @@ where
             Err(_) => return Ok(vec![]),
         };
 
-        if !fields.contains_key(&header) {
+        if !fields.as_ref().contains_key(&header) {
             return Ok(vec![]);
         }
 
         let res = fields
+            .as_ref()
             .get_all(&header)
             .into_iter()
             .map(|val| val.as_bytes().to_owned())
@@ -157,7 +168,7 @@ where
         let fields = get_fields(self.table(), &fields).context("[fields_get] getting fields")?;
 
         match hyper::header::HeaderName::from_bytes(name.as_bytes()) {
-            Ok(header) => Ok(fields.contains_key(&header)),
+            Ok(header) => Ok(fields.as_ref().contains_key(&header)),
             Err(_) => Ok(false),
         }
     }
@@ -185,14 +196,18 @@ where
             }
         }
 
-        Ok(get_fields_mut(self.table(), &fields)
+        match get_fields_mut(self.table(), &fields)
             .context("[fields_set] getting mutable fields")?
-            .map(|fields| {
-                fields.remove(&header);
+        {
+            Ok(fields) => {
+                fields.remove_all(&header);
                 for value in values {
-                    fields.append(&header, value);
+                    fields.append(&header, value)?;
                 }
-            }))
+                Ok(Ok(()))
+            }
+            Err(e) => Ok(Err(e)),
+        }
     }
 
     fn delete(
@@ -210,7 +225,7 @@ where
         }
 
         Ok(get_fields_mut(self.table(), &fields)?.map(|fields| {
-            fields.remove(header);
+            fields.remove_all(&header);
         }))
     }
 
@@ -234,18 +249,23 @@ where
             Err(_) => return Ok(Err(types::HeaderError::InvalidSyntax)),
         };
 
-        Ok(get_fields_mut(self.table(), &fields)
+        match get_fields_mut(self.table(), &fields)
             .context("[fields_append] getting mutable fields")?
-            .map(|fields| {
-                fields.append(header, value);
-            }))
+        {
+            Ok(fields) => {
+                fields.append(&header, value)?;
+                Ok(Ok(()))
+            }
+            Err(e) => Ok(Err(e)),
+        }
     }
 
     fn entries(
         &mut self,
         fields: Resource<HostFields>,
     ) -> wasmtime::Result<Vec<(String, Vec<u8>)>> {
         Ok(get_fields(self.table(), &fields)?
+            .as_ref()
             .iter()
             .map(|(name, value)| (name.as_str().to_owned(), value.as_bytes().to_owned()))
             .collect())
@@ -270,7 +290,7 @@ where
     T: WasiHttpView,
 {
     fn method(&mut self, id: Resource<HostIncomingRequest>) -> wasmtime::Result<Method> {
-        let method = self.table().get(&id)?.parts.method.clone();
+        let method = self.table().get(&id)?.method.clone();
         Ok(method.into())
     }
     fn path_with_query(
@@ -279,7 +299,6 @@ where
     ) -> wasmtime::Result<Option<String>> {
         let req = self.table().get(&id)?;
         Ok(req
-            .parts
             .uri
             .path_and_query()
             .map(|path_and_query| path_and_query.as_str().to_owned()))
@@ -300,11 +319,7 @@ where
         let _ = self.table().get(&id)?;
 
         fn get_fields(elem: &mut dyn Any) -> &mut FieldMap {
-            &mut elem
-                .downcast_mut::<HostIncomingRequest>()
-                .unwrap()
-                .parts
-                .headers
+            &mut elem.downcast_mut::<HostIncomingRequest>().unwrap().headers
         }
 
         let headers = self.table().push_child(
@@ -376,7 +391,7 @@ where
             return Ok(Err(()));
         }
 
-        let size = match get_content_length(&req.headers) {
+        let size = match get_content_length(req.headers.as_ref()) {
             Ok(size) => size,
             Err(..) => return Ok(Err(())),
         };
@@ -653,7 +668,7 @@ where
     {
         let trailers = self.table().get_mut(&id)?;
         match trailers {
-            HostFutureTrailers::Waiting(_) => return Ok(None),
+            HostFutureTrailers::Waiting { .. } => return Ok(None),
             HostFutureTrailers::Consumed => return Ok(Some(Err(()))),
             HostFutureTrailers::Done(_) => {}
         };
@@ -742,7 +757,7 @@ where
             return Ok(Err(()));
         }
 
-        let size = match get_content_length(&resp.headers) {
+        let size = match get_content_length(resp.headers.as_ref()) {
             Ok(size) => size,
             Err(..) => return Ok(Err(())),
         };
@@ -821,6 +836,7 @@ where
     ) -> wasmtime::Result<
         Option<Result<Result<Resource<HostIncomingResponse>, types::ErrorCode>, ()>>,
     > {
+        let field_size_limit = self.ctx().field_size_limit;
         let resp = self.table().get_mut(&id)?;
 
         match resp {
@@ -841,15 +857,17 @@ where
                 Ok(Err(e)) => return Ok(Some(Ok(Err(e)))),
             };
 
-        let (mut parts, body) = resp.resp.into_parts();
+        let (parts, body) = resp.resp.into_parts();
 
-        remove_forbidden_headers(self, &mut parts.headers);
+        let mut headers = FieldMap::new(parts.headers, field_size_limit);
+        remove_forbidden_headers(self, &mut headers);
 
         let resp = self.table().push(HostIncomingResponse {
             status: parts.status.as_u16(),
-            headers: parts.headers,
+            headers,
             body: Some({
-                let mut body = HostIncomingBody::new(body, resp.between_bytes_timeout);
+                let mut body =
+                    HostIncomingBody::new(body, resp.between_bytes_timeout, field_size_limit);
                 if let Some(worker) = resp.worker {
                     body.retain_worker(worker);
                 }

@@ -7,11 +7,13 @@ use crate::{
 };
 use anyhow::bail;
 use bytes::Bytes;
+use http::header::{HeaderMap, HeaderName, HeaderValue};
 use http_body_util::BodyExt;
 use hyper::body::Body;
-use hyper::header::HeaderName;
 use std::any::Any;
+use std::fmt;
 use std::time::Duration;
+use wasmtime::Result;
 use wasmtime::component::{Resource, ResourceTable};
 use wasmtime_wasi::p2::Pollable;
 use wasmtime_wasi::runtime::AbortOnDropJoinHandle;
@@ -24,16 +26,39 @@ use {
     tokio::time::timeout,
 };
 
+/// Default maximum size for the contents of a fields resource.
+///
+/// Typically, HTTP proxies limit headers to 8k. This number is higher than that
+/// because it not only includes the wire-size of headers but it additionally
+/// includes factors for the in-memory representation of `HeaderMap`. This is in
+/// theory high enough that no one runs into it but low enough such that a
+/// completely full `HeaderMap` doesn't break the bank in terms of memory
+/// consumption.
+const DEFAULT_FIELD_SIZE_LIMIT: usize = 2 << 30;
+
 /// Capture the state necessary for use in the wasi-http API implementation.
 #[derive(Debug)]
 pub struct WasiHttpCtx {
-    _priv: (),
+    pub(crate) field_size_limit: usize,
 }
 
 impl WasiHttpCtx {
     /// Create a new context.
     pub fn new() -> Self {
-        Self { _priv: () }
+        Self {
+            field_size_limit: DEFAULT_FIELD_SIZE_LIMIT,
+        }
+    }
+
+    /// Set the maximum size for any fields resources created by this context.
+    ///
+    /// The limit specified here is roughly a byte limit for the size of the
+    /// in-memory representation of headers. This means that the limit needs to
+    /// be larger than the literal representation of headers on the wire to
+    /// account for in-memory Rust-side data structures representing the header
+    /// names/values/etc.
+    pub fn set_field_size_limit(&mut self, limit: usize) {
+        self.field_size_limit = limit;
     }
 }
 
@@ -95,14 +120,17 @@ pub trait WasiHttpView {
         B: Body<Data = Bytes, Error = hyper::Error> + Send + 'static,
         Self: Sized,
     {
+        let field_size_limit = self.ctx().field_size_limit;
         let (parts, body) = req.into_parts();
         let body = body.map_err(crate::hyper_response_error).boxed_unsync();
         let body = HostIncomingBody::new(
             body,
             // TODO: this needs to be plumbed through
             std::time::Duration::from_millis(600 * 1000),
+            field_size_limit,
         );
-        let incoming_req = HostIncomingRequest::new(self, parts, scheme, Some(body))?;
+        let incoming_req =
+            HostIncomingRequest::new(self, parts, scheme, Some(body), field_size_limit)?;
         Ok(self.table().push(incoming_req)?)
     }
 
@@ -305,12 +333,9 @@ pub const DEFAULT_FORBIDDEN_HEADERS: [http::header::HeaderName; 9] = [
     HeaderName::from_static("http2-settings"),
 ];
 
-/// Removes forbidden headers from a [`hyper::HeaderMap`].
-pub(crate) fn remove_forbidden_headers(
-    view: &mut dyn WasiHttpView,
-    headers: &mut hyper::HeaderMap,
-) {
-    let forbidden_keys = Vec::from_iter(headers.keys().filter_map(|name| {
+/// Removes forbidden headers from a [`FieldMap`].
+pub(crate) fn remove_forbidden_headers(view: &mut dyn WasiHttpView, headers: &mut FieldMap) {
+    let forbidden_keys = Vec::from_iter(headers.as_ref().keys().filter_map(|name| {
         if view.is_forbidden_header(name) {
             Some(name.clone())
         } else {
@@ -319,7 +344,7 @@ pub(crate) fn remove_forbidden_headers(
     }));
 
     for name in forbidden_keys {
-        headers.remove(name);
+        headers.remove_all(&name);
     }
 }
 
@@ -533,7 +558,9 @@ impl TryInto<http::Method> for types::Method {
 /// The concrete type behind a `wasi:http/types.incoming-request` resource.
 #[derive(Debug)]
 pub struct HostIncomingRequest {
-    pub(crate) parts: http::request::Parts,
+    pub(crate) method: http::method::Method,
+    pub(crate) uri: http::uri::Uri,
+    pub(crate) headers: FieldMap,
     pub(crate) scheme: Scheme,
     pub(crate) authority: String,
     /// The body of the incoming request.
@@ -544,9 +571,10 @@ impl HostIncomingRequest {
     /// Create a new `HostIncomingRequest`.
     pub fn new(
         view: &mut dyn WasiHttpView,
-        mut parts: http::request::Parts,
+        parts: http::request::Parts,
         scheme: Scheme,
         body: Option<HostIncomingBody>,
+        field_size_limit: usize,
     ) -> anyhow::Result<Self> {
         let authority = match parts.uri.authority() {
             Some(authority) => authority.to_string(),
@@ -556,9 +584,13 @@ impl HostIncomingRequest {
             },
         };
 
-        remove_forbidden_headers(view, &mut parts.headers);
+        let mut headers = FieldMap::new(parts.headers, field_size_limit);
+        remove_forbidden_headers(view, &mut headers);
+
         Ok(Self {
-            parts,
+            method: parts.method,
+            uri: parts.uri,
+            headers,
             authority,
             scheme,
             body,
@@ -593,7 +625,7 @@ impl TryFrom<HostOutgoingResponse> for hyper::Response<HyperOutgoingBody> {
 
         let mut builder = hyper::Response::builder().status(resp.status);
 
-        *builder.headers_mut().unwrap() = resp.headers;
+        *builder.headers_mut().unwrap() = resp.headers.map;
 
         match resp.body {
             Some(body) => builder.body(body),
@@ -667,8 +699,130 @@ pub enum HostFields {
     },
 }
 
-/// An owned version of `HostFields`
-pub type FieldMap = hyper::HeaderMap;
+/// An owned version of `HostFields`. A wrapper on http `HeaderMap` that
+/// keeps a running tally of memory consumed by header names and values.
+#[derive(Debug, Clone)]
+pub struct FieldMap {
+    map: HeaderMap,
+    limit: usize,
+    size: usize,
+}
+
+/// Error given when a `FieldMap` has exceeded the size limit.
+#[derive(Debug)]
+pub struct FieldSizeLimitError {
+    /// The erroring `FieldMap` operation would require this content size
+    pub(crate) size: usize,
+    /// The limit set on `FieldMap` content size
+    pub(crate) limit: usize,
+}
+impl fmt::Display for FieldSizeLimitError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "Field size limit {} exceeded: {}", self.limit, self.size)
+    }
+}
+impl std::error::Error for FieldSizeLimitError {}
+
+impl FieldMap {
+    /// Construct a `FieldMap` from a `HeaderMap` and a size limit.
+    ///
+    /// Construction with a `HeaderMap` which exceeds the size limit is
+    /// allowed, but subsequent operations to expand the resource use will
+    /// fail.
+    pub fn new(map: HeaderMap, limit: usize) -> Self {
+        let size = Self::content_size(&map);
+        Self { map, size, limit }
+    }
+    /// Construct an empty `FieldMap`
+    pub fn empty(limit: usize) -> Self {
+        Self {
+            map: HeaderMap::new(),
+            size: 0,
+            limit,
+        }
+    }
+    /// Get the `HeaderMap` out of the `FieldMap`
+    pub fn into_inner(self) -> HeaderMap {
+        self.map
+    }
+    /// Calculate the content size of a `HeaderMap`. This is a sum of the size
+    /// of all of the keys and all of the values.
+    pub(crate) fn content_size(map: &HeaderMap) -> usize {
+        let mut sum = 0;
+        for key in map.keys() {
+            sum += header_name_size(key);
+        }
+        for value in map.values() {
+            sum += header_value_size(value);
+        }
+        sum
+    }
+    /// Remove all values associated with a key in a map.
+    ///
+    /// Returns an empty list if the key is not already present within the map.
+    pub fn remove_all(&mut self, key: &HeaderName) -> Vec<HeaderValue> {
+        use http::header::Entry;
+        match self.map.try_entry(key) {
+            Ok(Entry::Vacant { .. }) | Err(_) => Vec::new(),
+            Ok(Entry::Occupied(e)) => {
+                let (name, value_drain) = e.remove_entry_mult();
+                let mut removed = header_name_size(&name);
+                let values = value_drain.collect::<Vec<_>>();
+                for v in values.iter() {
+                    removed += header_value_size(v);
+                }
+                self.size -= removed;
+                values
+            }
+        }
+    }
+    /// Add a value associated with a key to the map.
+    ///
+    /// If `key` is already present within the map then `value` is appended to
+    /// the list of values it already has.
+    pub fn append(&mut self, key: &HeaderName, value: HeaderValue) -> Result<bool> {
+        let key_size = header_name_size(key);
+        let val_size = header_value_size(&value);
+        let new_size = if !self.map.contains_key(key) {
+            self.size + key_size + val_size
+        } else {
+            self.size + val_size
+        };
+        if new_size > self.limit {
+            bail!(FieldSizeLimitError {
+                limit: self.limit,
+                size: new_size
+            })
+        }
+        self.size = new_size;
+        Ok(self.map.try_append(key, value)?)
+    }
+}
+
+/// Returns the size, in accounting cost, to consider for `name`.
+///
+/// This includes both the byte length of the `name` itself as well as the size
+/// of the data structure itself as it'll reside within a `HeaderMap`.
+fn header_name_size(name: &HeaderName) -> usize {
+    name.as_str().len() + size_of::<HeaderName>()
+}
+
+/// Same as `header_name_size`, but for values.
+///
+/// This notably includes the size of `HeaderValue` itself to ensure that all
+/// headers have a nonzero size as otherwise this would never limit addition of
+/// an empty header value.
+fn header_value_size(value: &HeaderValue) -> usize {
+    value.len() + size_of::<HeaderValue>()
+}
+
+// We impl AsRef, but not AsMut, because any modifications of the
+// underlying HeaderMap must account for changes in size
+impl AsRef<HeaderMap> for FieldMap {
+    fn as_ref(&self) -> &HeaderMap {
+        &self.map
+    }
+}
 
 /// A handle to a future incoming response.
 pub type FutureIncomingResponseHandle =

@@ -123,6 +123,7 @@ async fn run_wasi_http(
     send_request: Option<RequestSender>,
     rejected_authority: Option<String>,
     early_drop: bool,
+    field_size_limit: Option<usize>,
 ) -> anyhow::Result<Result<hyper::Response<Collected<Bytes>>, ErrorCode>> {
     let stdout = MemoryOutputPipe::new(4096);
     let stderr = MemoryOutputPipe::new(4096);
@@ -140,7 +141,10 @@ async fn run_wasi_http(
     builder.stdout(stdout.clone());
     builder.stderr(stderr.clone());
     let wasi = builder.build();
-    let http = WasiHttpCtx::new();
+    let mut http = WasiHttpCtx::new();
+    if let Some(limit) = field_size_limit {
+        http.set_field_size_limit(limit);
+    }
     let ctx = Ctx {
         table,
         wasi,
@@ -153,15 +157,22 @@ async fn run_wasi_http(
     let mut store = Store::new(&engine, ctx);
 
     let mut linker = Linker::new(&engine);
-    wasmtime_wasi_http::add_to_linker_async(&mut linker)?;
+    wasmtime_wasi_http::add_to_linker_async(&mut linker).context("add crate to linker")?;
     let proxy =
         wasmtime_wasi_http::bindings::Proxy::instantiate_async(&mut store, &component, &linker)
-            .await?;
+            .await
+            .context("instantiate proxy")?;
 
-    let req = store.data_mut().new_incoming_request(Scheme::Http, req)?;
+    let req = store
+        .data_mut()
+        .new_incoming_request(Scheme::Http, req)
+        .context("new incoming request")?;
 
     let (sender, receiver) = tokio::sync::oneshot::channel();
-    let out = store.data_mut().new_response_outparam(sender)?;
+    let out = store
+        .data_mut()
+        .new_response_outparam(sender)
+        .context("new response outparam")?;
 
     let receiver = if early_drop {
         // Drop the receiver early, emulating a host event like
@@ -176,7 +187,8 @@ async fn run_wasi_http(
         proxy
             .wasi_http_incoming_handler()
             .call_handle(&mut store, req, out)
-            .await?;
+            .await
+            .context("calling incoming handler")?;
 
         Ok::<_, anyhow::Error>(())
     });
@@ -185,7 +197,7 @@ async fn run_wasi_http(
         let resp = match r.await {
             Ok(Ok(resp)) => {
                 let (parts, body) = resp.into_parts();
-                let collected = BodyExt::collect(body).await?;
+                let collected = BodyExt::collect(body).await.context("collecting body")?;
                 Some(Ok(hyper::Response::from_parts(parts, collected)))
             }
             Ok(Err(e)) => Some(Err(e)),
@@ -197,9 +209,9 @@ async fn run_wasi_http(
 
         // Now that the response has been processed, we can wait on the wasm to
         // finish without deadlocking.
-        handle.await.context("Component execution")?;
+        handle.await.context("awaiting execution")?;
 
-        Ok(resp.expect("wasm never called set-response-outparam"))
+        Ok(resp.context("wasm never called set-response-outparam")?)
     } else {
         handle.await.context("Component execution")?;
         Ok(Err(ErrorCode::HttpResponseTimeout))
@@ -219,6 +231,7 @@ async fn wasi_http_proxy_tests() -> anyhow::Result<()> {
         None,
         None,
         false,
+        None,
     )
     .await?;
 
@@ -351,6 +364,7 @@ async fn do_wasi_http_hash_all(override_send_request: bool) -> Result<()> {
         send_request,
         None,
         false,
+        None,
     )
     .await??;
 
@@ -399,6 +413,7 @@ async fn wasi_http_hash_all_with_reject() -> Result<()> {
         None,
         Some("forbidden.com".to_string()),
         false,
+        None,
     )
     .await??;
 
@@ -519,6 +534,7 @@ async fn do_wasi_http_echo(uri: &str, url_header: Option<&str>) -> Result<()> {
         None,
         None,
         false,
+        None,
     )
     .await??;
 
@@ -551,6 +567,7 @@ async fn wasi_http_without_port() -> Result<()> {
         None,
         None,
         false,
+        None,
     )
     .await??;
 
@@ -574,6 +591,7 @@ async fn wasi_http_no_trap_on_early_drop() -> Result<()> {
         None,
         None,
         true,
+        None,
     )
     .await?;
 
@@ -583,3 +601,98 @@ async fn wasi_http_no_trap_on_early_drop() -> Result<()> {
         panic!("test expects an error");
     }
 }
+
+#[test_log::test(tokio::test)]
+async fn wasi_http_fields_limit_incoming_request() -> Result<()> {
+    use crate::p2::types::FieldSizeLimitError;
+    use http::{HeaderName, HeaderValue, Request};
+    use http_body_util::combinators::BoxBody;
+    use hyper::Error;
+
+    fn request_with_header_size(uri: &str, total: usize) -> Request<BoxBody<Bytes, Error>> {
+        let mut builder = hyper::Request::builder().uri(uri).method(http::Method::GET);
+
+        let headers = builder.headers_mut().expect("builder error");
+        let chunks = total / 10;
+        let remainder = total % 10;
+        for chunk in 0..chunks {
+            let mut v = format!("v{chunk:04}");
+            if chunk == 0 {
+                for _ in 0..remainder {
+                    v.push('x');
+                }
+            }
+            headers.insert(
+                HeaderName::from_bytes(format!("k{chunk:04}").as_bytes())
+                    .expect("valid header name"),
+                HeaderValue::from_str(&v).expect("valid header value"),
+            );
+        }
+        builder
+            .body(body::empty())
+            .expect("complete building request")
+    }
+
+    let resp = run_wasi_http(
+        test_programs_artifacts::P2_API_PROXY_COMPONENT,
+        request_with_header_size("http://host/", 256),
+        None,
+        None,
+        false,
+        Some(255),
+    )
+    .await
+    .context("request with headers on wire over the size limit")??;
+    assert_eq!(resp.status(), 200);
+
+    let resp = run_wasi_http(
+        test_programs_artifacts::P2_API_PROXY_COMPONENT,
+        request_with_header_size("http://host/new_fields/50", 0),
+        None,
+        None,
+        false,
+        Some(500),
+    )
+    .await
+    .context("new_fields with size matching the size limit")??;
+    assert_eq!(resp.status(), 200);
+
+    let err = run_wasi_http(
+        test_programs_artifacts::P2_API_PROXY_COMPONENT,
+        request_with_header_size("http://host/new_fields/500", 0),
+        None,
+        None,
+        false,
+        Some(500),
+    )
+    .await
+    .err()
+    .expect("new_fields exceeding the size limit");
+    assert!(err.downcast_ref::<FieldSizeLimitError>().is_some());
+
+    let resp = run_wasi_http(
+        test_programs_artifacts::P2_API_PROXY_COMPONENT,
+        request_with_header_size("http://host/modify_fields/50", 10),
+        None,
+        None,
+        false,
+        Some(500),
+    )
+    .await??;
+    assert_eq!(resp.status(), 200);
+
+    let err = run_wasi_http(
+        test_programs_artifacts::P2_API_PROXY_COMPONENT,
+        request_with_header_size("http://host/modify_fields/50", 100),
+        None,
+        None,
+        false,
+        Some(500),
+    )
+    .await
+    .err()
+    .expect("run_wasi_http should give error");
+    assert!(err.downcast_ref::<FieldSizeLimitError>().is_some());
+
+    Ok(())
+}

@@ -5,6 +5,7 @@ use alloc::collections::BTreeMap;
 use alloc::string::String;
 use alloc::vec::Vec;
 use anyhow::{Result, anyhow};
+use bytes::Bytes;
 use core::future::Future;
 use core::pin::Pin;
 use core::task::{Context, Poll};
@@ -163,9 +164,10 @@ impl streams::HostOutputStream for ResourceTable {
             ));
         }
 
-        self.get_mut(&stream)?
-            .blocking_write_zeroes_and_flush(len as usize)
-            .await
+        // TODO: We could optimize this to not allocate one big zeroed buffer, and instead write
+        // repeatedly from a 'static buffer of zeros.
+        let bs = Bytes::from_iter(core::iter::repeat(0).take(len as usize));
+        self.get_mut(&stream)?.blocking_write_and_flush(bs).await
     }
 
     fn write_zeroes(&mut self, stream: Resource<DynOutputStream>, len: u64) -> StreamResult<()> {

@@ -233,44 +233,19 @@ pub trait OutputStream: Pollable {
     /// Returning an Err which downcasts to a [`StreamError`] will be
     /// reported to Wasm as the empty error result. Otherwise, errors will trap.
     fn write_zeroes(&mut self, nelem: usize) -> StreamResult<()> {
+        let n = self.check_write()?;
+        if nelem > n {
+            return Err(StreamError::trap(
+                "cannot write more zeroes than `check_write` allows",
+            ));
+        };
         // TODO: We could optimize this to not allocate one big zeroed buffer, and instead write
         // repeatedly from a 'static buffer of zeros.
         let bs = Bytes::from_iter(core::iter::repeat(0).take(nelem));
         self.write(bs)?;
         Ok(())
     }
 
-    /// Perform a write of up to 4096 zeroes, and then flush the stream.
-    /// Block until all of these operations are complete, or an error
-    /// occurs.
-    ///
-    /// This is a convenience wrapper around the use of `check-write`,
-    /// `subscribe`, `write-zeroes`, and `flush`, and is implemented with
-    /// the following pseudo-code:
-    ///
-    /// ```text
-    /// let pollable = this.subscribe();
-    /// while num_zeroes != 0 {
-    ///     // Wait for the stream to become writable
-    ///     pollable.block();
-    ///     let Ok(n) = this.check-write(); // eliding error handling
-    ///     let len = min(n, num_zeroes);
-    ///     this.write-zeroes(len);         // eliding error handling
-    ///     num_zeroes -= len;
-    /// }
-    /// this.flush();
-    /// // Wait for completion of `flush`
-    /// pollable.block();
-    /// // Check for any errors that arose during `flush`
-    /// let _ = this.check-write();         // eliding error handling
-    /// ```
-    async fn blocking_write_zeroes_and_flush(&mut self, nelem: usize) -> StreamResult<()> {
-        // TODO: We could optimize this to not allocate one big zeroed buffer, and instead write
-        // repeatedly from a 'static buffer of zeros.
-        let bs = Bytes::from_iter(core::iter::repeat(0).take(nelem));
-        self.blocking_write_and_flush(bs).await
-    }
-
     /// Simultaneously waits for this stream to be writable and then returns how
     /// much may be written or the last error that happened.
     async fn write_ready(&mut self) -> StreamResult<usize> {

@@ -120,7 +120,7 @@ impl Pollable for InputFile {
 
 impl InputStream for InputFile {
     fn read(&mut self, size: usize) -> StreamResult<Bytes> {
-        let mut buf = bytes::BytesMut::zeroed(size);
+        let mut buf = bytes::BytesMut::zeroed(size.min(crate::MAX_READ_SIZE_ALLOC));
         let bytes_read = self
             .file
             .read(&mut buf)

@@ -350,6 +350,20 @@ impl WasiCtxBuilder {
         self
     }
 
+    /// Configures the maximum len accepted by
+    /// `wasi:random/random.get-random-bytes` and
+    /// `wasi:random/insecure.get-insecure-random-bytes`. Calls with a len
+    /// larger than this limit will trap.
+    ///
+    /// This limit protects the host implementation from memory exhaustion from
+    /// untrusted guest input. A limit of `u64::MAX` is equivalent to no limit,
+    /// but note that this enables a guest to also force the host to attempt an
+    /// allocation of that size.
+    pub fn max_random_size(&mut self, max_size: u64) -> &mut Self {
+        self.random.max_size = max_size;
+        self
+    }
+
     /// Configures `wasi:clocks/wall-clock` to use the `clock` specified.
     ///
     /// By default the host's wall clock is used.

@@ -12,6 +12,24 @@
 //!
 //! For WASIp3, see [`p3`]. WASIp3 support is experimental, unstable and incomplete.
 
+/// The maximum size, in bytes, that this crate will allocate on the host on a
+/// per-read basis.
+///
+/// This is used to limit the size of `wasi:io/streams.input-stream#read`, for
+/// example, along with a variety of other read-style APIs. All of these APIs
+/// have the shape where the guest asks the host to read some data for it, and
+/// then it's returned. The data is allocated on the host, however, meaning that
+/// when the guest asks for an extremely large read that could cause a very
+/// large allocations on the host. For now this constant serves as a hard limit
+/// on the size of the allocation a host will make.
+///
+/// TODO: make this configurable per-I/O? Per-WASI? It's not great that this
+/// just a hard and unconfigurable limit. There are probably situations where
+/// performance will be improved if this limit were higher or removed. Doing so
+/// without exposing hosts to guest-controlled resource exhaustion is the
+/// important part, though.
+const MAX_READ_SIZE_ALLOC: usize = 64 * 1024;
+
 pub mod cli;
 pub mod clocks;
 mod ctx;

@@ -70,6 +70,9 @@ impl wiggle::GuestErrorType for types::Errno {
 }
 
 impl<T: Snapshot1 + Send> wasi_unstable::WasiUnstable for T {
+    fn set_hostcall_fuel(&mut self, fuel: usize) {
+        Snapshot1::set_hostcall_fuel(self, fuel)
+    }
     fn args_get(
         &mut self,
         memory: &mut GuestMemory<'_>,

@@ -145,6 +145,7 @@ pub struct WasiP1Ctx {
     table: ResourceTable,
     wasi: WasiCtx,
     adapter: WasiP1Adapter,
+    hostcall_fuel: usize,
 }
 
 impl WasiP1Ctx {
@@ -153,8 +154,81 @@ impl WasiP1Ctx {
             table: ResourceTable::new(),
             wasi,
             adapter: WasiP1Adapter::new(),
+            hostcall_fuel: 0,
         }
     }
+
+    fn consume_fuel(&mut self, fuel: usize) -> Result<()> {
+        if fuel > self.hostcall_fuel {
+            return Err(types::Errno::Nomem.into());
+        }
+        self.hostcall_fuel -= fuel;
+        Ok(())
+    }
+
+    /// Assumes the host is going to copy all of `array` in which case a
+    /// corresponding amount of fuel is consumed to ensure it's not too large.
+    fn consume_fuel_for_array<T>(&mut self, array: wiggle::GuestPtr<[T]>) -> Result<()> {
+        let byte_size = usize::try_from(array.len())?
+            .checked_mul(size_of::<T>())
+            .ok_or(types::Errno::Overflow)?;
+        self.consume_fuel(byte_size)
+    }
+
+    /// Returns the first non-empty buffer in `ciovs` or a single empty buffer
+    /// if they're all empty.
+    ///
+    /// Additionally consumes a corresponding amount of fuel appropriate to the
+    /// size of `ciovs` and the first non-empty array.
+    fn first_non_empty_ciovec(
+        &mut self,
+        memory: &GuestMemory<'_>,
+        ciovs: types::CiovecArray,
+    ) -> Result<GuestPtr<[u8]>> {
+        self.consume_fuel_for_array(ciovs)?;
+        for iov in ciovs.iter() {
+            let iov = memory.read(iov?)?;
+            if iov.buf_len == 0 {
+                continue;
+            }
+            let ret = iov.buf.as_array(iov.buf_len);
+            self.consume_fuel_for_array(ret)?;
+            return Ok(ret);
+        }
+        Ok(GuestPtr::new((0, 0)))
+    }
+
+    /// Returns the first non-empty buffer in `iovs` or a single empty buffer if
+    /// they're all empty.
+    ///
+    /// Additionally consumes a corresponding amount of fuel appropriate to the
+    /// size of `ciovs` and the first non-empty array.
+    fn first_non_empty_iovec(
+        &mut self,
+        memory: &GuestMemory<'_>,
+        iovs: types::IovecArray,
+    ) -> Result<GuestPtr<[u8]>> {
+        self.consume_fuel_for_array(iovs)?;
+        for iov in iovs.iter() {
+            let iov = memory.read(iov?)?;
+            if iov.buf_len == 0 {
+                continue;
+            }
+            let ret = iov.buf.as_array(iov.buf_len);
+            self.consume_fuel_for_array(ret)?;
+            return Ok(ret);
+        }
+        Ok(GuestPtr::new((0, 0)))
+    }
+
+    /// Copies the guest string `ptr` into the host.
+    ///
+    /// Consumes a corresponding amount of fuel for the byte size of `ptr` and
+    /// fails if it's too large.
+    fn read_string(&mut self, memory: &GuestMemory<'_>, ptr: GuestPtr<str>) -> Result<String> {
+        self.consume_fuel(usize::try_from(ptr.len())?)?;
+        Ok(memory.as_cow_str(ptr)?.into_owned())
+    }
 }
 
 impl WasiView for WasiP1Ctx {
@@ -558,6 +632,7 @@ impl WasiP1Ctx {
         ciovs: types::CiovecArray,
         write: FdWrite,
     ) -> Result<types::Size, types::Error> {
+        let buf = self.first_non_empty_ciovec(memory, ciovs)?;
         let t = self.transact()?;
         let desc = t.get_descriptor(fd)?;
         match desc {
@@ -579,7 +654,6 @@ impl WasiP1Ctx {
                 let append = *append;
                 drop(t);
                 let f = self.table.get(&fd)?.file()?;
-                let buf = first_non_empty_ciovec(memory, ciovs)?;
 
                 let do_write = move |f: &cap_std::fs::File, buf: &[u8]| match (append, write) {
                     // Note that this is implementing Linux semantics of
@@ -630,7 +704,6 @@ impl WasiP1Ctx {
                 }
                 let stream = stream.borrowed();
                 drop(t);
-                let buf = first_non_empty_ciovec(memory, ciovs)?;
                 let n = BlockingMode::Blocking
                     .write(memory, &mut self.table, stream, buf)
                     .await?
@@ -1124,46 +1197,14 @@ fn write_byte(memory: &mut GuestMemory<'_>, ptr: GuestPtr<u8>, byte: u8) -> Resu
     Ok(next)
 }
 
-fn read_string<'a>(memory: &'a GuestMemory<'_>, ptr: GuestPtr<str>) -> Result<String> {
-    Ok(memory.as_cow_str(ptr)?.into_owned())
-}
-
-// Returns the first non-empty buffer in `ciovs` or a single empty buffer if
-// they're all empty.
-fn first_non_empty_ciovec(
-    memory: &GuestMemory<'_>,
-    ciovs: types::CiovecArray,
-) -> Result<GuestPtr<[u8]>> {
-    for iov in ciovs.iter() {
-        let iov = memory.read(iov?)?;
-        if iov.buf_len == 0 {
-            continue;
-        }
-        return Ok(iov.buf.as_array(iov.buf_len));
-    }
-    Ok(GuestPtr::new((0, 0)))
-}
-
-// Returns the first non-empty buffer in `iovs` or a single empty buffer if
-// they're all empty.
-fn first_non_empty_iovec(
-    memory: &GuestMemory<'_>,
-    iovs: types::IovecArray,
-) -> Result<GuestPtr<[u8]>> {
-    for iov in iovs.iter() {
-        let iov = memory.read(iov?)?;
-        if iov.buf_len == 0 {
-            continue;
-        }
-        return Ok(iov.buf.as_array(iov.buf_len));
-    }
-    Ok(GuestPtr::new((0, 0)))
-}
-
 // Implement the WasiSnapshotPreview1 trait using only the traits that are
 // required for T, i.e., in terms of the preview 2 wit interface, and state
 // stored in the WasiP1Adapter struct.
 impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
+    fn set_hostcall_fuel(&mut self, fuel: usize) {
+        self.hostcall_fuel = fuel;
+    }
+
     #[instrument(skip(self, memory))]
     fn args_get(
         &mut self,
@@ -1646,6 +1687,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         fd: types::Fd,
         iovs: types::IovecArray,
     ) -> Result<types::Size, types::Error> {
+        let iov = self.first_non_empty_iovec(memory, iovs)?;
         let t = self.transact()?;
         let desc = t.get_descriptor(fd)?;
         match desc {
@@ -1662,7 +1704,6 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                 drop(t);
                 let pos = position.load(Ordering::Relaxed);
                 let file = self.table.get(&fd)?.file()?;
-                let iov = first_non_empty_iovec(memory, iovs)?;
                 let bytes_read = match (file.as_blocking_file(), memory.as_slice_mut(iov)?) {
                     // Try to read directly into wasm memory where possible
                     // when the current thread can block and additionally wasm
@@ -1701,15 +1742,14 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
             Descriptor::Stdin { stream, .. } => {
                 let stream = stream.borrowed();
                 drop(t);
-                let buf = first_non_empty_iovec(memory, iovs)?;
                 let read = BlockingMode::Blocking
-                    .read(&mut self.table, stream, buf.len().try_into()?)
+                    .read(&mut self.table, stream, iov.len().try_into()?)
                     .await?;
-                if read.len() > buf.len().try_into()? {
+                if read.len() > iov.len().try_into()? {
                     return Err(types::Errno::Range.into());
                 }
-                let buf = buf.get_range(0..u32::try_from(read.len())?).unwrap();
-                memory.copy_from_slice(&read, buf)?;
+                let iov = iov.get_range(0..u32::try_from(read.len())?).unwrap();
+                memory.copy_from_slice(&read, iov)?;
                 let n = read.len().try_into()?;
                 Ok(n)
             }
@@ -1727,6 +1767,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         iovs: types::IovecArray,
         offset: types::Filesize,
     ) -> Result<types::Size, types::Error> {
+        let buf = self.first_non_empty_iovec(memory, iovs)?;
         let t = self.transact()?;
         let desc = t.get_descriptor(fd)?;
         let (buf, read) = match desc {
@@ -1736,7 +1777,6 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                 let fd = fd.borrowed();
                 let blocking_mode = *blocking_mode;
                 drop(t);
-                let buf = first_non_empty_iovec(memory, iovs)?;
 
                 let stream = self.filesystem().read_via_stream(fd, offset)?;
                 let read = blocking_mode
@@ -2024,7 +2064,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         path: GuestPtr<str>,
     ) -> Result<(), types::Error> {
         let dirfd = self.get_dir_fd(dirfd)?;
-        let path = read_string(memory, path)?;
+        let path = self.read_string(memory, path)?;
         self.filesystem()
             .create_directory_at(dirfd.borrowed(), path)
             .await?;
@@ -2042,7 +2082,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         path: GuestPtr<str>,
     ) -> Result<types::Filestat, types::Error> {
         let dirfd = self.get_dir_fd(dirfd)?;
-        let path = read_string(memory, path)?;
+        let path = self.read_string(memory, path)?;
         let filesystem::DescriptorStat {
             type_,
             link_count: nlink,
@@ -2103,7 +2143,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         )?;
 
         let dirfd = self.get_dir_fd(dirfd)?;
-        let path = read_string(memory, path)?;
+        let path = self.read_string(memory, path)?;
         self.filesystem()
             .set_times_at(dirfd, flags.into(), path, atim, mtim)
             .await?;
@@ -2124,8 +2164,8 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
     ) -> Result<(), types::Error> {
         let src_fd = self.get_dir_fd(src_fd)?;
         let target_fd = self.get_dir_fd(target_fd)?;
-        let src_path = read_string(memory, src_path)?;
-        let target_path = read_string(memory, target_path)?;
+        let src_path = self.read_string(memory, src_path)?;
+        let target_path = self.read_string(memory, target_path)?;
         self.filesystem()
             .link_at(src_fd, src_flags.into(), src_path, target_fd, target_path)
             .await?;
@@ -2146,7 +2186,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         _fs_rights_inheriting: types::Rights,
         fdflags: types::Fdflags,
     ) -> Result<types::Fd, types::Error> {
-        let path = read_string(memory, path)?;
+        let path = self.read_string(memory, path)?;
 
         let mut flags = filesystem::DescriptorFlags::empty();
         if fs_rights_base.contains(types::Rights::FD_READ) {
@@ -2205,7 +2245,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         buf_len: types::Size,
     ) -> Result<types::Size, types::Error> {
         let dirfd = self.get_dir_fd(dirfd)?;
-        let path = read_string(memory, path)?;
+        let path = self.read_string(memory, path)?;
         let mut path = self
             .filesystem()
             .readlink_at(dirfd, path)
@@ -2228,7 +2268,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         path: GuestPtr<str>,
     ) -> Result<(), types::Error> {
         let dirfd = self.get_dir_fd(dirfd)?;
-        let path = read_string(memory, path)?;
+        let path = self.read_string(memory, path)?;
         self.filesystem().remove_directory_at(dirfd, path).await?;
         Ok(())
     }
@@ -2246,8 +2286,8 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
     ) -> Result<(), types::Error> {
         let src_fd = self.get_dir_fd(src_fd)?;
         let dest_fd = self.get_dir_fd(dest_fd)?;
-        let src_path = read_string(memory, src_path)?;
-        let dest_path = read_string(memory, dest_path)?;
+        let src_path = self.read_string(memory, src_path)?;
+        let dest_path = self.read_string(memory, dest_path)?;
         self.filesystem()
             .rename_at(src_fd, src_path, dest_fd, dest_path)
             .await?;
@@ -2263,8 +2303,8 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
         dest_path: GuestPtr<str>,
     ) -> Result<(), types::Error> {
         let dirfd = self.get_dir_fd(dirfd)?;
-        let src_path = read_string(memory, src_path)?;
-        let dest_path = read_string(memory, dest_path)?;
+        let src_path = self.read_string(memory, src_path)?;
+        let dest_path = self.read_string(memory, dest_path)?;
         self.filesystem()
             .symlink_at(dirfd.borrowed(), src_path, dest_path)
             .await?;
@@ -2332,9 +2372,18 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
 
         let subs = subs.as_array(nsubscriptions);
         let events = events.as_array(nsubscriptions);
-
         let n = usize::try_from(nsubscriptions).unwrap_or(usize::MAX);
-        let mut pollables = Vec::with_capacity(n);
+
+        self.consume_fuel_for_array(subs)?;
+        self.consume_fuel_for_array(events)?;
+
+        let mut temp = TempResources {
+            ctx: self,
+            pollables: Vec::with_capacity(n),
+            inputs: Vec::new(),
+            outputs: Vec::new(),
+        };
+        let mut borrowed_pollables = Vec::with_capacity(n);
         for sub in subs.iter() {
             let sub = memory.read(sub?)?;
             let p = match sub.u {
@@ -2349,7 +2398,7 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                         types::Clockid::Monotonic => (timeout, absolute),
                         types::Clockid::Realtime if !absolute => (timeout, false),
                         types::Clockid::Realtime => {
-                            let now = wall_clock::Host::now(&mut self.clocks())
+                            let now = wall_clock::Host::now(&mut temp.ctx.clocks())
                                 .context("failed to call `wall_clock::now`")
                                 .map_err(types::Error::trap)?;
 
@@ -2372,11 +2421,11 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                         _ => return Err(types::Errno::Inval.into()),
                     };
                     if absolute {
-                        monotonic_clock::Host::subscribe_instant(&mut self.clocks(), timeout)
+                        monotonic_clock::Host::subscribe_instant(&mut temp.ctx.clocks(), timeout)
                             .context("failed to call `monotonic_clock::subscribe_instant`")
                             .map_err(types::Error::trap)?
                     } else {
-                        monotonic_clock::Host::subscribe_duration(&mut self.clocks(), timeout)
+                        monotonic_clock::Host::subscribe_duration(&mut temp.ctx.clocks(), timeout)
                             .context("failed to call `monotonic_clock::subscribe_duration`")
                             .map_err(types::Error::trap)?
                     }
@@ -2385,29 +2434,32 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                     file_descriptor,
                 }) => {
                     let stream = {
-                        let t = self.transact()?;
+                        let t = temp.ctx.transact()?;
                         let desc = t.get_descriptor(file_descriptor)?;
                         match desc {
                             Descriptor::Stdin { stream, .. } => stream.borrowed(),
                             Descriptor::File(File { fd, position, .. }) => {
                                 let pos = position.load(Ordering::Relaxed);
                                 let fd = fd.borrowed();
                                 drop(t);
-                                self.filesystem().read_via_stream(fd, pos)?
+                                let r = temp.ctx.filesystem().read_via_stream(fd, pos)?;
+                                let ret = r.borrowed();
+                                temp.inputs.push(r);
+                                ret
                             }
                             // TODO: Support sockets
                             _ => return Err(types::Errno::Badf.into()),
                         }
                     };
-                    streams::HostInputStream::subscribe(&mut self.table, stream)
+                    streams::HostInputStream::subscribe(&mut temp.ctx.table, stream)
                         .context("failed to call `subscribe` on `input-stream`")
                         .map_err(types::Error::trap)?
                 }
                 types::SubscriptionU::FdWrite(types::SubscriptionFdReadwrite {
                     file_descriptor,
                 }) => {
                     let stream = {
-                        let t = self.transact()?;
+                        let t = temp.ctx.transact()?;
                         let desc = t.get_descriptor(file_descriptor)?;
                         match desc {
                             Descriptor::Stdout { stream, .. }
@@ -2422,32 +2474,38 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                                 let position = position.clone();
                                 let append = *append;
                                 drop(t);
-                                if append {
-                                    self.filesystem().append_via_stream(fd)?
+                                let r = if append {
+                                    temp.ctx.filesystem().append_via_stream(fd)?
                                 } else {
                                     let pos = position.load(Ordering::Relaxed);
-                                    self.filesystem().write_via_stream(fd, pos)?
-                                }
+                                    temp.ctx.filesystem().write_via_stream(fd, pos)?
+                                };
+                                let ret = r.borrowed();
+                                temp.outputs.push(r);
+                                ret
                             }
                             // TODO: Support sockets
                             _ => return Err(types::Errno::Badf.into()),
                         }
                     };
-                    streams::HostOutputStream::subscribe(&mut self.table, stream)
+                    streams::HostOutputStream::subscribe(&mut temp.ctx.table, stream)
                         .context("failed to call `subscribe` on `output-stream`")
                         .map_err(types::Error::trap)?
                 }
             };
-            pollables.push(p);
+            borrowed_pollables.push(p.borrowed());
+            temp.pollables.push(p);
         }
-        let ready: HashSet<_> = self
+        let ready: HashSet<_> = temp
+            .ctx
             .table
-            .poll(pollables)
+            .poll(borrowed_pollables)
             .await
             .context("failed to call `poll-oneoff`")
             .map_err(types::Error::trap)?
             .into_iter()
             .collect();
+        drop(temp);
 
         let mut count: types::Size = 0;
         for (sub, event) in (0..)
@@ -2544,7 +2602,31 @@ impl wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiP1Ctx {
                 .checked_add(1)
                 .ok_or_else(|| types::Error::from(types::Errno::Overflow))?
         }
-        Ok(count)
+        return Ok(count);
+
+        struct TempResources<'a> {
+            ctx: &'a mut WasiP1Ctx,
+            pollables: Vec<Resource<crate::p2::bindings::io::streams::Pollable>>,
+            inputs: Vec<Resource<crate::p2::bindings::io::streams::InputStream>>,
+            outputs: Vec<Resource<crate::p2::bindings::io::streams::OutputStream>>,
+        }
+
+        impl Drop for TempResources<'_> {
+            fn drop(&mut self) {
+                for p in self.pollables.drain(..) {
+                    use wasmtime_wasi_io::bindings::wasi::io::poll::HostPollable;
+                    self.ctx.table.drop(p).unwrap();
+                }
+                for p in self.inputs.drain(..) {
+                    assert!(p.owned());
+                    self.ctx.table.delete(p).unwrap();
+                }
+                for p in self.outputs.drain(..) {
+                    assert!(p.owned());
+                    self.ctx.table.delete(p).unwrap();
+                }
+            }
+        }
     }
 
     #[instrument(skip(self, _memory))]

@@ -86,7 +86,7 @@ impl FileInputStream {
     fn blocking_read(file: &cap_std::fs::File, offset: u64, size: usize) -> ReadState {
         use system_interface::fs::FileIoExt;
 
-        let mut buf = BytesMut::zeroed(size);
+        let mut buf = BytesMut::zeroed(size.min(crate::MAX_READ_SIZE_ALLOC));
         loop {
             match file.read_at(&mut buf, offset) {
                 Ok(0) => return ReadState::Closed,

@@ -115,7 +115,12 @@ impl HostDescriptor for WasiFilesystemCtxView<'_> {
 
         let (mut buffer, r) = f
             .run_blocking(move |f| {
-                let mut buffer = vec![0; len.try_into().unwrap_or(usize::MAX)];
+                let mut buffer = vec![
+                    0;
+                    len.try_into()
+                        .unwrap_or(usize::MAX)
+                        .min(crate::MAX_READ_SIZE_ALLOC)
+                ];
                 let r = f.read_vectored_at(&mut [IoSliceMut::new(&mut buffer)], offset);
                 (buffer, r)
             })

@@ -1,9 +1,13 @@
 use crate::p2::bindings::random::{insecure, insecure_seed, random};
 use crate::random::WasiRandomCtx;
+use anyhow::bail;
 use cap_rand::{Rng, distributions::Standard};
 
 impl random::Host for WasiRandomCtx {
     fn get_random_bytes(&mut self, len: u64) -> anyhow::Result<Vec<u8>> {
+        if len > self.max_size {
+            bail!("requested len {len:?} exceeds limit {}", self.max_size);
+        }
         Ok((&mut self.random)
             .sample_iter(Standard)
             .take(len as usize)
@@ -17,6 +21,9 @@ impl random::Host for WasiRandomCtx {
 
 impl insecure::Host for WasiRandomCtx {
     fn get_insecure_random_bytes(&mut self, len: u64) -> anyhow::Result<Vec<u8>> {
+        if len > self.max_size {
+            bail!("requested len {len:?} exceeds limit {}", self.max_size);
+        }
         Ok((&mut self.insecure_random)
             .sample_iter(Standard)
             .take(len as usize)

@@ -216,10 +216,12 @@ impl udp::HostIncomingDatagramStream for WasiSocketsCtxView<'_> {
         }
 
         let mut datagrams = vec![];
+        let mut sum = 0;
 
-        while datagrams.len() < max_results {
+        while datagrams.len() < max_results && sum < crate::MAX_READ_SIZE_ALLOC {
             match recv_one(stream) {
                 Ok(Some(datagram)) => {
+                    sum += 1 + datagram.data.len();
                     datagrams.push(datagram);
                 }
                 Ok(None) => {

@@ -163,7 +163,7 @@ impl AsyncReadStream {
             let mut reader = pin!(reader);
             loop {
                 use tokio::io::AsyncReadExt;
-                let mut buf = bytes::BytesMut::with_capacity(4096);
+                let mut buf = bytes::BytesMut::with_capacity(crate::MAX_READ_SIZE_ALLOC);
                 let sent = match reader.read_buf(&mut buf).await {
                     Ok(nbytes) if nbytes == 0 => sender.send(Err(StreamError::Closed)).await,
                     Ok(_) => sender.send(Ok(buf.freeze())).await,
@@ -516,29 +516,31 @@ mod test {
     // suitable design for all applications, and we will probably make a knob or change the
     // behavior at some point, but this test shows the behavior as it is implemented:
     async fn backpressure_read_stream() {
-        let (r, mut w) = simplex(16 * 1024); // Make sure this buffer isn't a bottleneck
+        let (r, mut w) = simplex(4 * crate::MAX_READ_SIZE_ALLOC); // Make sure this buffer isn't a bottleneck
         let mut reader = AsyncReadStream::new(r);
 
         let writer_task = tokio::task::spawn(async move {
             // Write twice as much as we can buffer up in an AsyncReadStream:
-            w.write_all(&[123; 8192]).await.unwrap();
+            w.write_all(&[123; 2 * crate::MAX_READ_SIZE_ALLOC])
+                .await
+                .unwrap();
             w
         });
 
         resolves_immediately(reader.ready()).await;
 
         // Now we expect the reader task has sent 4k from the stream to the reader.
         // Try to read out one bigger than the buffer available:
-        let bs = reader.read(4097).unwrap();
-        assert_eq!(bs.len(), 4096);
+        let bs = reader.read(crate::MAX_READ_SIZE_ALLOC + 1).unwrap();
+        assert_eq!(bs.len(), crate::MAX_READ_SIZE_ALLOC);
 
         // Allow the crank to turn more:
         resolves_immediately(reader.ready()).await;
 
         // Again we expect the reader task has sent 4k from the stream to the reader.
         // Try to read out one bigger than the buffer available:
-        let bs = reader.read(4097).unwrap();
-        assert_eq!(bs.len(), 4096);
+        let bs = reader.read(crate::MAX_READ_SIZE_ALLOC + 1).unwrap();
+        assert_eq!(bs.len(), crate::MAX_READ_SIZE_ALLOC);
 
         // The writer task is now finished - join with it:
         let w = resolves_immediately(writer_task).await;
@@ -550,7 +552,10 @@ mod test {
         resolves_immediately(reader.ready()).await;
 
         // Now we expect the reader to be empty, and the stream.dropd:
-        assert!(matches!(reader.read(4097), Err(StreamError::Closed)));
+        assert!(matches!(
+            reader.read(crate::MAX_READ_SIZE_ALLOC + 1),
+            Err(StreamError::Closed)
+        ));
     }
 
     #[test_log::test(test_log::test(tokio::test(flavor = "multi_thread")))]

@@ -69,7 +69,7 @@ impl TcpReader {
             return Ok(bytes::Bytes::new());
         }
 
-        let mut buf = bytes::BytesMut::with_capacity(size);
+        let mut buf = bytes::BytesMut::with_capacity(size.min(crate::MAX_READ_SIZE_ALLOC));
         let n = match self.stream.try_read_buf(&mut buf) {
             // A 0-byte read indicates that the stream has closed.
             Ok(0) => {

@@ -41,10 +41,15 @@ impl HasData for WasiRandom {
     type Data<'a> = &'a mut WasiRandomCtx;
 }
 
+/// Default largest length accepted by wasi 0.2 `get-random-bytes` and
+/// `get-insecure-random-bytes` methods.
+pub const DEFAULT_MAX_SIZE: u64 = 2 << 30;
+
 pub struct WasiRandomCtx {
     pub(crate) random: Box<dyn RngCore + Send>,
     pub(crate) insecure_random: Box<dyn RngCore + Send>,
     pub(crate) insecure_random_seed: u128,
+    pub(crate) max_size: u64,
 }
 
 impl Default for WasiRandomCtx {
@@ -60,10 +65,12 @@ impl Default for WasiRandomCtx {
         // API.
         let insecure_random_seed =
             cap_rand::thread_rng(cap_rand::ambient_authority()).r#gen::<u128>();
+        let max_size = DEFAULT_MAX_SIZE;
         Self {
             random: thread_rng(),
             insecure_random,
             insecure_random_seed,
+            max_size,
         }
     }
 }

@@ -3,6 +3,7 @@ use anyhow::Result;
 use std::path::Path;
 use test_programs_artifacts::*;
 use wasmtime::{Linker, Module};
+use wasmtime_wasi::WasiView;
 use wasmtime_wasi::p1::{WasiP1Ctx, add_to_linker_async};
 
 async fn run(path: &str, inherit_stdio: bool) -> Result<()> {
@@ -21,6 +22,7 @@ async fn run(path: &str, inherit_stdio: bool) -> Result<()> {
         }
         builder.build_p1()
     })?;
+    store.data_mut().wasi.ctx().table.set_max_capacity(1000);
     let instance = linker.instantiate_async(&mut store, &module).await?;
     let start = instance.get_typed_func::<(), ()>(&mut store, "_start")?;
     start.call_async(&mut store, ()).await?;
@@ -243,9 +245,7 @@ async fn p1_file_write() {
 async fn p1_path_open_lots() {
     run(P1_PATH_OPEN_LOTS, true).await.unwrap()
 }
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p1_cli_much_stdout() {}
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
+async fn p1_sleep_quickly_but_lots() {
+    run(P1_SLEEP_QUICKLY_BUT_LOTS, true).await.unwrap()
+}

@@ -19,10 +19,7 @@ async fn run(path: &str, inherit_stdio: bool) -> Result<()> {
         if inherit_stdio {
             builder.inherit_stdio();
         }
-        MyWasiCtx {
-            wasi: builder.build(),
-            table: Default::default(),
-        }
+        MyWasiCtx::new(builder.build())
     })?;
     let component = Component::from_file(&engine, path)?;
     let command = Command::instantiate_async(&mut store, &component, &linker).await?;
@@ -262,6 +259,12 @@ async fn p1_file_write() {
 async fn p1_path_open_lots() {
     run(P1_PATH_OPEN_LOTS_COMPONENT, false).await.unwrap()
 }
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
+async fn p1_sleep_quickly_but_lots() {
+    run(P1_SLEEP_QUICKLY_BUT_LOTS_COMPONENT, false)
+        .await
+        .unwrap()
+}
 
 #[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn p2_sleep() {
@@ -359,8 +362,3 @@ async fn p2_adapter_badfd() {
 async fn p2_file_read_write() {
     run(P2_FILE_READ_WRITE_COMPONENT, false).await.unwrap()
 }
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p1_cli_much_stdout() {}

@@ -21,10 +21,7 @@ fn run(path: &str, inherit_stdio: bool) -> Result<()> {
                 builder.inherit_stdio();
             }
             builder.allow_blocking_current_thread(blocking);
-            MyWasiCtx {
-                wasi: builder.build(),
-                table: Default::default(),
-            }
+            MyWasiCtx::new(builder.build())
         })?;
         let command = Command::instantiate(&mut store, &component, &linker)?;
         command
@@ -250,6 +247,10 @@ fn p1_file_write() {
 fn p1_path_open_lots() {
     run(P1_PATH_OPEN_LOTS_COMPONENT, false).unwrap()
 }
+#[test_log::test]
+fn p1_sleep_quickly_but_lots() {
+    run(P1_SLEEP_QUICKLY_BUT_LOTS_COMPONENT, false).unwrap()
+}
 
 #[test_log::test]
 fn p2_sleep() {
@@ -339,8 +340,3 @@ fn p2_adapter_badfd() {
 fn p2_file_read_write() {
     run(P2_FILE_READ_WRITE_COMPONENT, false).unwrap()
 }
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p1_cli_much_stdout() {}

@@ -26,11 +26,12 @@ async fn run_allow_blocking_current_thread(
         .context("failed to link `wasi:cli@0.2.x`")?;
     wasmtime_wasi::p3::add_to_linker(&mut linker).context("failed to link `wasi:cli@0.3.x`")?;
 
-    let (mut store, _td) = Ctx::new(&engine, name, |builder| MyWasiCtx {
-        wasi: builder
-            .allow_blocking_current_thread(allow_blocking_current_thread)
-            .build(),
-        table: Default::default(),
+    let (mut store, _td) = Ctx::new(&engine, name, |builder| {
+        MyWasiCtx::new(
+            builder
+                .allow_blocking_current_thread(allow_blocking_current_thread)
+                .build(),
+        )
     })?;
     let component = Component::from_file(&engine, path)?;
     let command = Command::instantiate_async(&mut store, &component, &linker)
@@ -151,33 +152,3 @@ async fn p3_file_write() -> anyhow::Result<()> {
 async fn p3_file_write_blocking() -> anyhow::Result<()> {
     run_allow_blocking_current_thread(P3_FILE_WRITE_COMPONENT, true).await
 }
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p3_cli_hello_stdout() {}
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p3_cli_hello_stdout_post_return() {}
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p3_cli_much_stdout() {}
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p3_cli_serve_hello_world() {}
-
-#[expect(
-    dead_code,
-    reason = "tested in the wasi-cli crate, satisfying foreach_api! macro"
-)]
-fn p3_cli_serve_sleep() {}

@@ -72,8 +72,16 @@ impl<T> Drop for Ctx<T> {
 }
 
 pub struct MyWasiCtx {
-    pub wasi: WasiCtx,
-    pub table: ResourceTable,
+    wasi: WasiCtx,
+    table: ResourceTable,
+}
+
+impl MyWasiCtx {
+    pub fn new(wasi: WasiCtx) -> Self {
+        let mut table = ResourceTable::new();
+        table.set_max_capacity(1000);
+        Self { wasi, table }
+    }
 }
 
 impl WasiView for Ctx<MyWasiCtx> {

@@ -5061,13 +5061,11 @@ impl TaskId {
     /// and delete the task when all threads are done.
     pub(crate) fn host_future_dropped<T>(&self, store: StoreContextMut<T>) -> Result<()> {
         let task = store.0.concurrent_state_mut().get_mut(self.task)?;
-        if !task.already_lowered_parameters() {
-            Waitable::Guest(self.task).delete_from(store.0.concurrent_state_mut())?
-        } else {
+        if task.already_lowered_parameters() {
             task.host_future_state = HostFutureState::Dropped;
-            if task.ready_to_delete() {
-                Waitable::Guest(self.task).delete_from(store.0.concurrent_state_mut())?
-            }
+        }
+        if task.ready_to_delete() {
+            Waitable::Guest(self.task).delete_from(store.0.concurrent_state_mut())?
         }
         Ok(())
     }

@@ -334,6 +334,12 @@ pub struct LiftContext<'a> {
         allow(unused, reason = "easier to not #[cfg] away")
     )]
     concurrent_state: &'a mut ConcurrentState,
+
+    /// Remaining fuel for this hostcall/lift operation.
+    ///
+    /// This is decremented for strings/lists, for example, to cap the size of
+    /// data the host allocates on behalf of the guest.
+    hostcall_fuel: usize,
 }
 
 #[doc(hidden)]
@@ -346,6 +352,7 @@ impl<'a> LiftContext<'a> {
         instance_handle: Instance,
     ) -> LiftContext<'a> {
         let store_id = store.id();
+        let hostcall_fuel = store.hostcall_fuel();
         // From `&mut StoreOpaque` provided the goal here is to project out
         // three different disjoint fields owned by the store: memory,
         // `CallContexts`, and `HandleTable`. There's no native API for that
@@ -369,6 +376,7 @@ impl<'a> LiftContext<'a> {
             host_table,
             host_resource_data,
             concurrent_state,
+            hostcall_fuel,
         }
     }
 
@@ -490,4 +498,21 @@ impl<'a> LiftContext<'a> {
     pub fn exit_call(&mut self) -> Result<()> {
         self.resource_tables().exit_call()
     }
+
+    /// Consumes `amt` units of fuel, typically a number of bytes, from this
+    /// context.
+    ///
+    /// Returns an error if the fuel is exhausted which will cause a trap in the
+    /// guest. Note that this is distinct from Wasm's fuel, this is just for
+    /// keeping track of data flowing from the guest to the host.
+    pub fn consume_fuel(&mut self, amt: usize) -> Result<()> {
+        match self.hostcall_fuel.checked_sub(amt) {
+            Some(new) => self.hostcall_fuel = new,
+            None => bail!(
+                "too much data is being copied between the host and the guest: \
+                 fuel allocated for hostcalls has been exhausted"
+            ),
+        }
+        Ok(())
+    }
 }

@@ -1637,7 +1637,7 @@ impl WasmStr {
             }
         };
         match byte_len.and_then(|len| ptr.checked_add(len)) {
-            Some(n) if n <= cx.memory().len() => {}
+            Some(n) if n <= cx.memory().len() => cx.consume_fuel(n - ptr)?,
             _ => bail!("string pointer/length out of bounds of memory"),
         }
         Ok(WasmStr {
@@ -1895,7 +1895,7 @@ impl<T: Lift> WasmList<T> {
             .checked_mul(T::SIZE32)
             .and_then(|len| ptr.checked_add(len))
         {
-            Some(n) if n <= cx.memory().len() => {}
+            Some(n) if n <= cx.memory().len() => cx.consume_fuel(n - ptr)?,
             _ => bail!("list pointer/length out of bounds of memory"),
         }
         if ptr % usize::try_from(T::ALIGN32)? != 0 {

@@ -36,6 +36,7 @@ impl core::error::Error for ResourceTableError {}
 pub struct ResourceTable {
     entries: Vec<Entry>,
     free_head: Option<usize>,
+    max_capacity: usize,
 }
 
 #[derive(Debug)]
@@ -66,6 +67,11 @@ struct Tombstone;
 // necessary.
 const DELETE_WITH_TOMBSTONE: bool = false;
 
+/// Default setting for `ResourceTable::max_capacity`, chosen to be high
+/// enough that it doesn't need changing all that often but low enough that
+/// exhausting it isn't a massive problem for the host.
+const DEFAULT_MAX_CAPACITY: usize = 1_000_000;
+
 /// This structure tracks parent and child relationships for a given table entry.
 ///
 /// Parents and children are referred to by table index. We maintain the
@@ -105,10 +111,7 @@ impl TableEntry {
 impl ResourceTable {
     /// Create an empty table
     pub fn new() -> Self {
-        ResourceTable {
-            entries: Vec::new(),
-            free_head: None,
-        }
+        ResourceTable::with_capacity(0)
     }
 
     /// Returns whether or not this table is empty.
@@ -122,11 +125,28 @@ impl ResourceTable {
         })
     }
 
+    /// Returns the maximum capacity of this table, in elements, before adding
+    /// any more will be refused.
+    pub fn max_capacity(&self) -> usize {
+        self.max_capacity
+    }
+
+    /// Configures the maximum number of entries that may be present within this
+    /// table.
+    ///
+    /// Note that this does not retroactively shrink the table nor evict
+    /// existing entries should the maximum be smaller than the current size of
+    /// the entry table.
+    pub fn set_max_capacity(&mut self, max: usize) {
+        self.max_capacity = max;
+    }
+
     /// Create an empty table with at least the specified capacity.
     pub fn with_capacity(capacity: usize) -> Self {
         ResourceTable {
             entries: Vec::with_capacity(capacity),
             free_head: None,
+            max_capacity: DEFAULT_MAX_CAPACITY,
         }
     }
 
@@ -197,6 +217,9 @@ impl ResourceTable {
             self.entries[free] = Entry::Occupied { entry: e };
             Ok(free.try_into().unwrap())
         } else {
+            if self.entries.len() >= self.max_capacity {
+                return Err(ResourceTableError::Full);
+            }
             let ix = self
                 .entries
                 .len()
@@ -446,3 +469,34 @@ pub fn test_free_list() {
     let x = table.push(()).unwrap();
     assert_eq!(x.rep(), 2);
 }
+
+#[test]
+fn test_max_capacity() {
+    let mut table = ResourceTable::new();
+    assert_eq!(table.max_capacity(), DEFAULT_MAX_CAPACITY);
+
+    table.set_max_capacity(0);
+    assert_eq!(table.max_capacity(), 0);
+    assert!(table.push(()).is_err());
+
+    table.set_max_capacity(1);
+    assert_eq!(table.max_capacity(), 1);
+    let x = table.push(()).unwrap();
+    assert!(table.push(()).is_err());
+
+    table.set_max_capacity(0);
+    assert!(table.push(()).is_err());
+    table.delete(x).unwrap();
+    let x = table.push(()).unwrap();
+    table.delete(x).unwrap();
+
+    table.set_max_capacity(10);
+
+    let handles = (0..10).map(|_| table.push(()).unwrap()).collect::<Vec<_>>();
+    assert!(table.push(()).is_err());
+    for handle in handles {
+        table.delete(handle).unwrap();
+    }
+
+    table.push(()).unwrap();
+}

@@ -2,14 +2,38 @@
 use crate::runtime::vm::VMStore;
 use crate::runtime::vm::component::{ComponentInstance, OwnedComponentInstance};
 use crate::store::{StoreData, StoreId, StoreOpaque};
+use crate::{AsContext, AsContextMut, Store, StoreContextMut};
 #[cfg(feature = "component-model-async")]
 use alloc::vec::Vec;
 use core::pin::Pin;
 use wasmtime_environ::PrimaryMap;
 
-#[derive(Default)]
+/// Default amount of fuel allowed for all guest-to-host calls in the component
+/// model.
+///
+/// This is the maximal amount of data which will be copied from the guest to
+/// the host by default. This is set to a very large value to avoid breaking
+/// existing embeddings for when this feature was backported.
+const DEFAULT_HOSTCALL_FUEL: usize = 2 << 30;
+
 pub struct ComponentStoreData {
     instances: PrimaryMap<ComponentInstanceId, Option<OwnedComponentInstance>>,
+
+    /// Fuel to be used for each time the guest calls the host or transfers data
+    /// to the host.
+    ///
+    /// Caps the size of the allocations made on the host to this amount
+    /// effectively.
+    hostcall_fuel: usize,
+}
+
+impl Default for ComponentStoreData {
+    fn default() -> Self {
+        Self {
+            instances: PrimaryMap::default(),
+            hostcall_fuel: DEFAULT_HOSTCALL_FUEL,
+        }
+    }
 }
 
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
@@ -85,6 +109,14 @@ impl StoreOpaque {
     pub(crate) fn component_instance(&self, id: ComponentInstanceId) -> &ComponentInstance {
         self.store_data().component_instance(id)
     }
+
+    pub(crate) fn hostcall_fuel(&self) -> usize {
+        self.store_data().components.hostcall_fuel
+    }
+
+    pub(crate) fn set_hostcall_fuel(&mut self, fuel: usize) {
+        self.store_data_mut().components.hostcall_fuel = fuel;
+    }
 }
 
 /// A type used to represent an allocated `ComponentInstance` located within a
@@ -174,3 +206,55 @@ impl StoreComponentInstanceId {
         store.component_instance_mut(self.instance)
     }
 }
+
+impl<T> Store<T> {
+    /// Returns the amount of "hostcall fuel" used for guest-to-host component
+    /// calls.
+    ///
+    /// This is either the default amount if it hasn't been configured or
+    /// returns the last value passed to [`Store::set_hostcall_fuel`].
+    ///
+    /// See [`Store::set_hostcall_fuel`] `for more details.
+    pub fn hostcall_fuel(&self) -> usize {
+        self.as_context().0.hostcall_fuel()
+    }
+
+    /// Sets the amount of "hostcall fuel" used for guest-to-host component
+    /// calls.
+    ///
+    /// Whenever the guest calls the host it often wants to transfer some data
+    /// as well, such as strings or lists. This configured fuel value can be
+    /// used to limit the amount of data that the host allocates on behalf of
+    /// the guest. This is a DoS mitigation mechanism to prevent a malicious
+    /// guest from causing the host to allocate an unbounded amount of memory
+    /// for example.
+    ///
+    /// Fuel is considered distinct for each host call. The host is responsible
+    /// for ensuring it retains a proper amount of data between host calls if
+    /// applicable. The `fuel` provided here will be the initial value for each
+    /// time the guest calls the host.
+    ///
+    /// The `fuel` value here should roughly corresponds to the maximal number
+    /// of bytes that the guest may transfer to the host in one call.
+    ///
+    /// Note that data transferred from the host to the guest is not limited
+    /// because it's already resident on the host itself. Only data from the
+    /// guest to the host is limited.
+    ///
+    /// The default value for this is 128 MiB.
+    pub fn set_hostcall_fuel(&mut self, fuel: usize) {
+        self.as_context_mut().set_hostcall_fuel(fuel)
+    }
+}
+
+impl<T> StoreContextMut<'_, T> {
+    /// See [`Store::hostcall_fuel`].
+    pub fn hostcall_fuel(&self) -> usize {
+        self.0.hostcall_fuel()
+    }
+
+    /// See [`Store::set_hostcall_fuel`].
+    pub fn set_hostcall_fuel(&mut self, fuel: usize) {
+        self.0.set_hostcall_fuel(fuel)
+    }
+}

@@ -915,7 +915,7 @@ fn load_list(cx: &mut LiftContext<'_>, ty: TypeListIndex, ptr: usize, len: usize
         .checked_mul(element_size)
         .and_then(|len| ptr.checked_add(len))
     {
-        Some(n) if n <= cx.memory().len() => {}
+        Some(n) if n <= cx.memory().len() => cx.consume_fuel(n - ptr)?,
         _ => bail!("list pointer/length out of bounds of memory"),
     }
     if ptr % usize::try_from(element_alignment)? != 0 {

@@ -81,6 +81,13 @@ pub fn define_module_trait(m: &Module, settings: &CodegenSettings) -> TokenStrea
     quote! {
         pub trait #traitname {
             #(#traitmethods)*
+
+            /// Indicates, if this implementation supports it, that the
+            /// specified amount of fuel should be the maximal spent for the
+            /// upcoming function call.
+            fn set_hostcall_fuel(&mut self, fuel: usize) {
+                let _ = fuel;
+            }
         }
     }
 }

@@ -116,14 +116,17 @@ fn generate_func(
 
     let body = quote! {
         let export = caller.get_export("memory");
+        let fuel = wiggle::wasmtime_crate::AsContextMut::as_context_mut(&mut caller).hostcall_fuel();
         let (mut mem, ctx) = match &export {
             Some(wiggle::wasmtime_crate::Extern::Memory(m)) => {
                 let (mem, ctx) = m.data_and_store_mut(&mut caller);
                 let ctx = get_cx(ctx);
+                ctx.set_hostcall_fuel(fuel);
                 (wiggle::GuestMemory::Unshared(mem), ctx)
             }
             Some(wiggle::wasmtime_crate::Extern::SharedMemory(m)) => {
                 let ctx = get_cx(caller.data_mut());
+                ctx.set_hostcall_fuel(fuel);
                 (wiggle::GuestMemory::Shared(m.data()), ctx)
             }
             _ => wiggle::anyhow::bail!("missing required memory export"),

@@ -1,3 +1,32 @@
+## 41.0.4
+
+Released 2026-02-24.
+
+### Changed
+
+* Wasmtime's implementation of WASI now has the ability to limit resource
+  consumption on behalf of the guest, such as host-allocated memory. This means
+  that some behaviors previously allowed by Wasmtime can now disallowed, such as
+  transferring excessive data from the guest to the host. Additionally calls to
+  `wasi:random/random.get-random-bytes`, for example, can have limits in place
+  to avoid allocating too much memory on the host. To preserve
+  backwards-compatible behavior these limits are NOT set by default. Embedders
+  must opt-in to configuring these knobs as appropriate for their embeddings.
+  For more information on this see the related security advisory with further
+  details on knobs added and what behaviors can be restricted.
+  [GHSA-852m-cvvp-9p4w](https://github.com/bytecodealliance/wasmtime/security/advisories/GHSA-852m-cvvp-9p4w)
+
+### Fixed
+
+* Panics when adding too many headers to a `wasi:http/types.fields` has been
+  resolved
+  [GHSA-243v-98vx-264h](https://github.com/bytecodealliance/wasmtime/security/advisories/GHSA-243v-98vx-264h)
+
+* Panic when dropping a `[Typed]Func::call_async` future.
+  [GHSA-xjhv-v822-pf94](https://github.com/bytecodealliance/wasmtime/security/advisories/GHSA-xjhv-v822-pf94)
+
+--------------------------------------------------------------------------------
+
 ## 41.0.3
 
 Released 2026-02-04.

@@ -1039,8 +1039,8 @@ impl RunCommand {
                         }
                     }
                 }
-
-                store.data_mut().wasi_http = Some(Arc::new(WasiHttpCtx::new()));
+                let http = self.run.wasi_http_ctx()?;
+                store.data_mut().wasi_http = Some(Arc::new(http));
             }
         }
 
@@ -1135,7 +1135,13 @@ impl RunCommand {
         let mut builder = wasmtime_wasi::WasiCtxBuilder::new();
         builder.inherit_stdio().args(&self.compute_argv()?);
         self.run.configure_wasip2(&mut builder)?;
-        let ctx = builder.build_p1();
+        let mut ctx = builder.build_p1();
+        if let Some(max) = self.run.common.wasi.max_resources {
+            ctx.ctx().table.set_max_capacity(max);
+        }
+        if let Some(fuel) = self.run.common.wasi.hostcall_fuel {
+            store.set_hostcall_fuel(fuel);
+        }
         store.data_mut().wasip1_ctx = Some(Arc::new(Mutex::new(ctx)));
         Ok(())
     }

@@ -230,10 +230,14 @@ impl ServeCommand {
         builder.stdout(LogStream::new(stdout_prefix, Output::Stdout));
         builder.stderr(LogStream::new(stderr_prefix, Output::Stderr));
 
+        let mut table = wasmtime::component::ResourceTable::new();
+        if let Some(max) = self.run.common.wasi.max_resources {
+            table.set_max_capacity(max);
+        }
         let mut host = Host {
-            table: wasmtime::component::ResourceTable::new(),
+            table,
             ctx: builder.build(),
-            http: WasiHttpCtx::new(),
+            http: self.run.wasi_http_ctx()?,
             http_outgoing_body_buffer_chunks: self.run.common.wasi.http_outgoing_body_buffer_chunks,
             http_outgoing_body_chunk_size: self.run.common.wasi.http_outgoing_body_chunk_size,
 
@@ -301,6 +305,10 @@ impl ServeCommand {
 
         let mut store = Store::new(engine, host);
 
+        if let Some(fuel) = self.run.common.wasi.hostcall_fuel {
+            store.set_hostcall_fuel(fuel);
+        }
+
         store.data_mut().limits = self.run.store_limits();
         store.limiter(|t| &mut t.limits);
 

@@ -323,10 +323,22 @@ impl RunCommon {
         if let Some(enable) = self.common.wasi.udp {
             builder.allow_udp(enable);
         }
+        if let Some(max_size) = self.common.wasi.max_random_size {
+            builder.max_random_size(max_size);
+        }
 
         Ok(())
     }
 
+    #[cfg(feature = "wasi-http")]
+    pub fn wasi_http_ctx(&self) -> Result<wasmtime_wasi_http::WasiHttpCtx> {
+        let mut http = wasmtime_wasi_http::WasiHttpCtx::new();
+        if let Some(limit) = self.common.wasi.max_http_fields_size {
+            http.set_field_size_limit(limit);
+        }
+        Ok(http)
+    }
+
     pub fn compute_preopen_sockets(&self) -> Result<Vec<TcpListener>> {
         let mut listeners = vec![];
 

@@ -172,9 +172,12 @@ fn assert_trap_code(status: &ExitStatus) {
 // Run a simple WASI hello world, snapshot0 edition.
 #[test]
 fn hello_wasi_snapshot0() -> Result<()> {
-    let wasm = build_wasm("tests/all/cli_tests/hello_wasi_snapshot0.wat")?;
     for preview2 in ["-Spreview2=n", "-Spreview2=y"] {
-        let stdout = run_wasmtime(&["-Ccache=n", preview2, wasm.path().to_str().unwrap()])?;
+        let stdout = run_wasmtime(&[
+            "-Ccache=n",
+            preview2,
+            "tests/all/cli_tests/hello_wasi_snapshot0.wat",
+        ])?;
         assert_eq!(stdout, "Hello, world!\n");
     }
     Ok(())
@@ -183,8 +186,7 @@ fn hello_wasi_snapshot0() -> Result<()> {
 // Run a simple WASI hello world, snapshot1 edition.
 #[test]
 fn hello_wasi_snapshot1() -> Result<()> {
-    let wasm = build_wasm("tests/all/cli_tests/hello_wasi_snapshot1.wat")?;
-    let stdout = run_wasmtime(&["-Ccache=n", wasm.path().to_str().unwrap()])?;
+    let stdout = run_wasmtime(&["-Ccache=n", "tests/all/cli_tests/hello_wasi_snapshot1.wat"])?;
     assert_eq!(stdout, "Hello, world!\n");
     Ok(())
 }
@@ -1104,7 +1106,7 @@ fn increase_stack_size() -> Result<()> {
 
 mod test_programs {
     use super::{get_wasmtime_command, run_wasmtime};
-    use anyhow::{Context, Result, bail};
+    use anyhow::{Context, Result, bail, format_err};
     use http_body_util::BodyExt;
     use hyper::header::HeaderValue;
     use std::io::{self, BufRead, BufReader, Read, Write};
@@ -1121,7 +1123,7 @@ mod test_programs {
             use self::$name as _;
         };
     }
-    foreach_p2_cli!(assert_test_exists);
+    foreach_cli!(assert_test_exists);
 
     #[test]
     fn p2_cli_hello_stdout() -> Result<()> {
@@ -2149,7 +2151,7 @@ start a print 1234
     }
 
     #[test]
-    fn p2_cli_p3_hello_stdout() -> Result<()> {
+    fn p3_cli_hello_stdout() -> Result<()> {
         let output = run_wasmtime(&[
             "run",
             "-Wcomponent-model-async",
@@ -2166,7 +2168,23 @@ start a print 1234
     }
 
     #[test]
-    fn p2_cli_p3_hello_stdout_post_return() -> Result<()> {
+    fn p2_cli_hello_stdout_invoke() -> Result<()> {
+        println!("{P2_CLI_HELLO_STDOUT_COMPONENT}");
+        let output = run_wasmtime(&[
+            "run",
+            "-Wcomponent-model",
+            "--invoke",
+            "run()",
+            P2_CLI_HELLO_STDOUT_COMPONENT,
+        ])?;
+        // First this component prints "hello, world", then the invoke
+        // result is printed as "ok".
+        assert_eq!(output, "hello, world\nok\n");
+        Ok(())
+    }
+
+    #[test]
+    fn p3_cli_hello_stdout_post_return() -> Result<()> {
         let output = run_wasmtime(&[
             "run",
             "-Wcomponent-model-async",
@@ -2182,24 +2200,130 @@ start a print 1234
         Ok(())
     }
 
-    mod invoke {
-        use super::*;
+    #[test]
+    fn p2_random_limits() -> Result<()> {
+        println!("{P2_RANDOM_COMPONENT}");
+
+        // By default, p2_random.rs fits within the random limits - always
+        // asks for 256 bytes.
+        let output = run_wasmtime(&["run", P2_RANDOM_COMPONENT])?;
+        assert_eq!(output, "");
+
+        // Lowering limit to 255 bytes should produce a trap in the first
+        // call, which goes by way of the wasip1 adapter:
+        let output = run_wasmtime(&["run", "-Smax-random-size=255", P2_RANDOM_COMPONENT])
+            .err()
+            .ok_or_else(|| format_err!("execution with max-random-size=255 should trap"))?;
+        let output = format!("{output:?}");
+        assert!(
+            output.contains("lib_generated::random_get"),
+            "expected error stack frames to contain 'wasip1::lib_generated::random_get'. Got:\n{output}"
+        );
+        assert!(
+            output.contains("requested len 256 exceeds limit 255"),
+            "expected error stack frames to contain 'requested len 256 exceeds limit 255'. Got:\n{output}"
+        );
+
+        // Lowering limit to 255 bytes and setting environment variable for
+        // the first call to only request 255 bytes should be OK, and then the
+        // program produce a trap in the second call, which calls the wasip2
+        // random import directly:
+        let output = run_wasmtime(&[
+            "run",
+            "-Smax-random-size=255",
+            "--env",
+            "TEST_P1_RANDOM_LEN=255",
+            P2_RANDOM_COMPONENT,
+        ])
+        .err()
+        .ok_or_else(|| format_err!("execution with max-random-size=255 should trap"))?;
+        let output = format!("{output:?}");
+        assert!(
+            output.contains("random::random::get_random_bytes"),
+            "expected error stack frames to contain 'wasi::random::random::get_random_bytes'. Got:\n{output}"
+        );
+        assert!(
+            output.contains("requested len 256 exceeds limit 255"),
+            "expected error stack frames to contain 'requested len 256 exceeds limit 255'. Got:\n{output}"
+        );
+
+        // Lowering limit to 255 bytes and setting environment variable for
+        // the first and second calls to be at the limit should produce a trap
+        // in the third call, which calls the wasip2 insecure random import:
+        let output = run_wasmtime(&[
+            "run",
+            "-Smax-random-size=255",
+            "--env",
+            "TEST_P1_RANDOM_LEN=255",
+            "--env",
+            "TEST_P2_RANDOM_LEN=255",
+            P2_RANDOM_COMPONENT,
+        ])
+        .err()
+        .ok_or_else(|| format_err!("execution with max-random-size=255 should trap"))?;
+        let output = format!("{output:?}");
+        assert!(
+            output.contains("random::insecure::get_insecure_random_bytes"),
+            "expected error stack frames to contain 'wasi::random::insecure::get_insecure_random_bytes'. Got:\n{output}"
+        );
+        assert!(
+            output.contains("requested len 256 exceeds limit 255"),
+            "expected error stack frames to contain 'requested len 256 exceeds limit 255'. Got:\n{output}"
+        );
+
+        // Lowering limit to 255 bytes and setting environment variable for
+        // the first and second calls to be under the limit should trap in the
+        // third call, which calls the wasip2 insecure random import:
+        let output = run_wasmtime(&[
+            "run",
+            "-Smax-random-size=255",
+            "--env",
+            "TEST_P1_RANDOM_LEN=255",
+            "--env",
+            "TEST_P2_RANDOM_LEN=255",
+            "--env",
+            "TEST_P2_INSECURE_RANDOM_LEN=255",
+            P2_RANDOM_COMPONENT,
+        ])
+        .context("setting all calls to be equal to the limit should pass")?;
+        assert_eq!(output, "");
+
+        Ok(())
+    }
 
-        #[test]
-        fn p2_cli_hello_stdout() -> Result<()> {
-            println!("{P2_CLI_HELLO_STDOUT_COMPONENT}");
-            let output = run_wasmtime(&[
+    #[test]
+    fn p2_cli_http_headers() -> Result<()> {
+        for test in ["append", "append-empty", "append-same", "append-same-empty"] {
+            let err = run_wasmtime(&[
                 "run",
-                "-Wcomponent-model",
-                "--invoke",
-                "run()",
-                P2_CLI_HELLO_STDOUT_COMPONENT,
-            ])?;
-            // First this component prints "hello, world", then the invoke
-            // result is printed as "ok".
-            assert_eq!(output, "hello, world\nok\n");
-            Ok(())
+                "-Shttp",
+                "-Smax-http-fields-size=1048576",
+                P2_CLI_HTTP_HEADERS_COMPONENT,
+                test,
+            ])
+            .unwrap_err();
+            assert!(
+                err.to_string()
+                    .contains("Field size limit 1048576 exceeded")
+                    || err.to_string().contains("max size reached"),
+                "bad error message: {err:?}"
+            );
         }
+
+        // With an extremely large limit Wasmtime still shouldn't panic.
+        let err = run_wasmtime(&[
+            "run",
+            "-Shttp",
+            &format!("-Smax-http-fields-size={}", 1 << 30),
+            P2_CLI_HTTP_HEADERS_COMPONENT,
+            "append",
+        ])
+        .unwrap_err();
+        assert!(
+            err.to_string().contains("max size reached"),
+            "bad error message: {err:?}"
+        );
+        Ok(())
     }
 
     #[test]
@@ -2262,6 +2386,18 @@ start a print 1234
         )
     }
 
+    #[test]
+    fn p2_cli_max_resources() -> Result<()> {
+        let err = run_wasmtime(&["run", "-Smax-resources=50", P2_CLI_MAX_RESOURCES_COMPONENT])
+            .unwrap_err();
+        assert!(
+            err.to_string().contains("resource table has no free keys"),
+            "bad error message: {err}"
+        );
+        run_wasmtime(&["run", "-Smax-resources=200", P2_CLI_MAX_RESOURCES_COMPONENT])?;
+        Ok(())
+    }
+
     #[tokio::test]
     #[cfg_attr(not(feature = "component-model-async"), ignore)]
     async fn p3_cli_serve_hello_world() -> Result<()> {
@@ -2502,6 +2638,98 @@ start a print 1234
         assert!(stderr.contains("guest timed out"), "bad stderr: {stderr}");
         Ok(())
     }
+
+    #[test]
+    fn p2_cli_many_resources() -> Result<()> {
+        let err = run_wasmtime(&["run", P2_CLI_MANY_RESOURCES_COMPONENT]).unwrap_err();
+        assert!(
+            err.to_string().contains("resource table has no free keys"),
+            "bad error message: {err}"
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn p1_cli_hostcall_fuel() -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        run_wasmtime(&[
+            "run",
+            &format!("--dir={}::.", dir.path().to_str().unwrap()),
+            "-Shostcall-fuel=1000",
+            P1_CLI_HOSTCALL_FUEL,
+        ])?;
+        Ok(())
+    }
+
+    #[test]
+    fn p2_cli_hostcall_fuel() -> Result<()> {
+        enum Exit {
+            Ok,
+            NoFuel,
+            TooManyZeroes,
+            BufferTooLarge,
+        }
+
+        let dir = tempfile::tempdir()?;
+        let file = dir.path().join("1mb");
+        std::fs::write(&file, vec![0; 1024 * 1024])?;
+        for (arg, exit) in [
+            ("poll", Exit::NoFuel),
+            ("read", Exit::Ok),
+            ("write", Exit::NoFuel),
+            ("mkdir", Exit::NoFuel),
+            ("write-stream", Exit::NoFuel),
+            ("write-stream-blocking", Exit::NoFuel),
+            ("resolve", Exit::NoFuel),
+            ("udp-send-many", Exit::NoFuel),
+            ("udp-send-big", Exit::NoFuel),
+            ("write-zeroes", Exit::TooManyZeroes),
+            ("write-stream-buffer-too-large", Exit::BufferTooLarge),
+            ("write-zeroes-buffer-too-large", Exit::BufferTooLarge),
+            ("read-file-big", Exit::Ok),
+            ("read-tcp-big", Exit::Ok),
+        ] {
+            println!("test: {arg}");
+            let result = run_wasmtime(&[
+                "run",
+                "-Shostcall-fuel=5000",
+                "-Sinherit-network",
+                &format!("--dir={}::.", dir.path().to_str().unwrap()),
+                P2_CLI_HOSTCALL_FUEL_COMPONENT,
+                arg,
+            ]);
+
+            match exit {
+                Exit::Ok => {
+                    result.unwrap();
+                }
+                Exit::NoFuel => {
+                    let err = result.unwrap_err();
+                    assert!(
+                        err.to_string()
+                            .contains("fuel allocated for hostcalls has been exhausted"),
+                        "bad error message: {err}"
+                    );
+                }
+                Exit::TooManyZeroes => {
+                    let err = result.unwrap_err();
+                    assert!(
+                        err.to_string()
+                            .contains("cannot write more zeroes than `check_write` allows"),
+                        "bad error message: {err}"
+                    );
+                }
+                Exit::BufferTooLarge => {
+                    let err = result.unwrap_err();
+                    assert!(
+                        err.to_string().contains("Buffer too large"),
+                        "bad error message: {err}"
+                    );
+                }
+            }
+        }
+        Ok(())
+    }
 }
 
 #[test]

@@ -3316,3 +3316,51 @@ fn run_export_with_internal_adapter() -> Result<()> {
     assert_eq!(run.call(&mut store, ())?, (5,));
     Ok(())
 }
+
+#[tokio::test]
+async fn drop_call_async_future() -> Result<()> {
+    let component = r#"
+(component
+  (import "foo" (func $f))
+  (core module $m
+    (func $f (import "" "foo"))
+    (func (export "foo") call $f)
+  )
+  (core func $f (canon lower (func $f)))
+  (core instance $m (instantiate $m (with "" (instance
+     (export "foo" (func $f))
+  ))))
+  (func (export "foo") (canon lift (core func $m "foo")))
+)
+"#;
+
+    let mut config = Config::new();
+    config.async_support(true);
+    let engine = &Engine::new(&config)?;
+    let component = Component::new(&engine, component)?;
+    let mut store = Store::new(&engine, ());
+    let mut linker = Linker::new(&engine);
+    linker.root().func_wrap_async("foo", |_, _: ()| {
+        Box::new(async {
+            tokio::task::yield_now().await;
+            Ok(())
+        })
+    })?;
+    let instance = linker.instantiate_async(&mut store, &component).await?;
+    let foo = instance.get_typed_func::<(), ()>(&mut store, "foo")?;
+    // Here we'll use `call_async` a few times but only poll each returned
+    // future once.  This will put the instance in a weird state but shouldn't
+    // cause a panic.
+    for _ in 0..5 {
+        let mut future = std::pin::pin!(async {
+            foo.call_async(&mut store, ()).await?;
+            foo.post_return_async(&mut store).await
+        });
+        if let std::task::Poll::Ready(result) =
+            std::future::poll_fn(|cx| std::task::Poll::Ready(future.as_mut().poll(cx))).await
+        {
+            _ = result;
+        }
+    }
+    Ok(())
+}

@@ -1203,3 +1203,47 @@ fn use_types_across_component_boundaries() -> Result<()> {
 
     Ok(())
 }
+
+#[test]
+fn hostcall_fuel_limits_val() -> Result<()> {
+    let engine = super::engine();
+    let component = Component::new(
+        &engine,
+        r#"(component
+            (import "hi" (func $hi (param "x" (list u8))))
+            (core module $libc
+                (memory (export "memory") 10)
+            )
+            (core module $m
+                (import "libc" "memory" (memory 1))
+                (import "" "hi" (func $hi (param i32 i32)))
+
+                (func (export "run")
+                    i32.const 0
+                    memory.size
+                    i32.const 65536
+                    i32.mul
+                    call $hi)
+            )
+            (core instance $libc (instantiate $libc))
+            (core func $hi (canon lower (func $hi) (memory $libc "memory")))
+            (core instance $i (instantiate $m
+                (with "libc" (instance $libc))
+                (with "" (instance (export "hi" (func $hi))))
+            ))
+            (func (export "run") (canon lift (core func $i "run")))
+        )"#,
+    )?;
+    let mut store = Store::new(&engine, 0);
+    let mut linker = Linker::new(&engine);
+    linker.root().func_new("hi", |_, _, _, _| Ok(()))?;
+    let instance = linker.instantiate(&mut store, &component)?;
+    let run = instance.get_func(&mut store, "run").unwrap();
+    run.call(&mut store, &[], &mut [])?;
+    run.post_return(&mut store)?;
+
+    store.set_hostcall_fuel(100);
+    assert!(run.call(&mut store, &[], &mut []).is_err());
+
+    Ok(())
+}