Use an in-memory FTP recovery transport

Cargo.lock

@@ -14,12 +14,6 @@ version = "2.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06"
 
-[[package]]
-name = "bytes"
-version = "1.12.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
-
 [[package]]
 name = "equivalent"
 version = "1.0.2"
@@ -77,12 +71,6 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2"
 
-[[package]]
-name = "libc"
-version = "0.2.189"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
-
 [[package]]
 name = "log"
 version = "0.4.34"
@@ -121,17 +109,6 @@ dependencies = [
  "wit-bindgen",
 ]
 
-[[package]]
-name = "mio"
-version = "1.2.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8"
-dependencies = [
- "libc",
- "wasi",
- "windows-sys",
-]
-
 [[package]]
 name = "pin-project-lite"
 version = "0.2.17"
@@ -214,16 +191,6 @@ dependencies = [
  "zmij",
 ]
 
-[[package]]
-name = "socket2"
-version = "0.6.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4"
-dependencies = [
- "libc",
- "windows-sys",
-]
-
 [[package]]
 name = "syn"
 version = "3.0.6"
@@ -241,13 +208,8 @@ version = "1.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
 dependencies = [
- "bytes",
- "libc",
- "mio",
  "pin-project-lite",
- "socket2",
  "tokio-macros",
- "windows-sys",
 ]
 
 [[package]]
@@ -267,12 +229,6 @@ version = "1.0.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954"
 
-[[package]]
-name = "wasi"
-version = "0.11.1+wasi-snapshot-preview1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
-
 [[package]]
 name = "wasm-encoder"
 version = "0.259.0"
@@ -307,21 +263,6 @@ dependencies = [
  "semver",
 ]
 
-[[package]]
-name = "windows-link"
-version = "0.2.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
-
-[[package]]
-name = "windows-sys"
-version = "0.61.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
-dependencies = [
- "windows-link",
-]
-
 [[package]]
 name = "wit-bindgen"
 version = "0.62.0"

Cargo.toml

@@ -11,7 +11,7 @@ crate-type = ["cdylib", "rlib"]
 memona-plugin-sdk = { path = "sdk" }
 
 [dev-dependencies]
-tokio = { version = "1", features = ["macros", "rt", "net", "io-util", "sync", "time"] }
+tokio = { version = "1", features = ["macros", "rt"] }
 
 [lints]
 workspace = true

src/session_tests.rs

@@ -1,25 +1,22 @@
-//! Real FTP commands against an async local server. The generated guest API is
+//! FTP command recovery against an in-memory protocol transport. Live QA
-//! synchronous, so each guest call is isolated with `spawn_blocking`; production
+//! separately exercises the compiled WASM component over real host sockets.
-//! WASM host integration is exercised separately.
 
-use super::Ftp;
+use super::{Ftp, test_host};
 use memona_plugin_sdk::{ErrorCode, Guest, Locator, Setting};
-use std::time::Duration;
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
-use tokio::net::{TcpListener, TcpStream};
-use tokio::sync::oneshot;
 
 type TestResult<T = ()> = Result<T, Box<dyn std::error::Error + Send + Sync>>;
-type Channel = BufReader<TcpStream>;
 
 fn guest<T>(value: Result<T, ErrorCode>) -> TestResult<T> {
   value.map_err(|error| format!("guest operation failed: {error:?}").into())
 }
 
-fn config(port: u16) -> Vec<(String, Setting)> {
+fn config() -> Vec<(String, Setting)> {
   vec![
     ("host".into(), Setting::Text("127.0.0.1".into())),
-    ("port".into(), Setting::Text(port.to_string())),
+    (
+      "port".into(),
+      Setting::Text(test_host::CONTROL_PORT.to_string()),
+    ),
     ("username".into(), Setting::Text("fixture".into())),
     (
       "password".into(),
@@ -28,140 +25,36 @@ fn config(port: u16) -> Vec<(String, Setting)> {
   ]
 }
 
-async fn reply(channel: &mut Channel, text: &str) -> TestResult {
-  channel.get_mut().write_all(text.as_bytes()).await?;
-  Ok(())
-}
-
-async fn expect(channel: &mut Channel, expected: &str) -> TestResult {
-  loop {
-    let mut line = String::new();
-    channel.read_line(&mut line).await?;
-    if line == "NOOP\r\n" {
-      reply(channel, "200 Still connected\r\n").await?;
-      continue;
-    }
-    assert_eq!(line, format!("{expected}\r\n"));
-    return Ok(());
-  }
-}
-
-async fn login(listener: &TcpListener) -> TestResult<Channel> {
-  let (stream, _) = listener.accept().await?;
-  let mut channel = BufReader::new(stream);
-  reply(&mut channel, "220 Fixture\r\n").await?;
-  expect(&mut channel, "USER fixture").await?;
-  reply(&mut channel, "331 Password required\r\n").await?;
-  expect(&mut channel, "PASS  fixture secret ").await?;
-  reply(&mut channel, "230 Signed in\r\n").await?;
-  expect(&mut channel, "TYPE I").await?;
-  reply(&mut channel, "200 Binary\r\n").await?;
-  Ok(channel)
-}
-
-async fn listing(channel: &mut Channel) -> TestResult {
-  expect(channel, "EPSV").await?;
-  let listener = TcpListener::bind("127.0.0.1:0").await?;
-  reply(
-    channel,
-    &format!("229 Passive (|||{}|)\r\n", listener.local_addr()?.port()),
-  )
-  .await?;
-  expect(channel, "MLSD /").await?;
-  reply(channel, "150 Listing\r\n").await?;
-  let (mut data, _) = listener.accept().await?;
-  data.write_all(b"type=file;size=3;  report.md \r\n").await?;
-  data.shutdown().await?;
-  drop(data);
-  reply(channel, "226 Complete\r\n").await
-}
-
 #[tokio::test]
 async fn a_listing_reconnects_after_the_server_expires_the_control_session()
 -> TestResult {
-  let listener = TcpListener::bind("127.0.0.1:0").await?;
+  let fixture = test_host::install(test_host::Scenario::ExpiredListing);
-  let port = listener.local_addr()?.port();
+  guest(Ftp::open("ftp".into(), config()))?;
-  let (expired, wait_expired) = oneshot::channel();
+  let first = guest(Ftp::list(Locator::Path("/".into()), None, 10))?;
-  let server = tokio::spawn(async move {
+  assert_eq!(first.entries[0].name, " report.md ");
-    tokio::time::timeout(Duration::from_secs(5), async {
+  let next = guest(Ftp::list(Locator::Path("/".into()), None, 10))?;
-      let mut channel = login(&listener).await?;
+  assert_eq!(next.entries[0].name, " report.md ");
-      listing(&mut channel).await?;
+  Ftp::close();
-      reply(&mut channel, "421 Control connection timed out\r\n").await?;
+  let summary = fixture.summary();
-      channel.get_mut().shutdown().await?;
+  assert_eq!(summary.control_connections, 2);
-      drop(channel);
+  assert_eq!(summary.listings, 2);
-      expired.send(()).map_err(|_| "guest disappeared")?;
+  assert_eq!(summary.mkdir_commands, 0);
-      let mut channel = login(&listener).await?;
-      listing(&mut channel).await?;
-      expect(&mut channel, "QUIT").await?;
-      reply(&mut channel, "221 Bye\r\n").await
-    })
-    .await?
-  });
-  // `Guest` is generated as a synchronous trait, so the repository rule for
-  // unavoidable blocking I/O requires isolating this native test call here.
-  let client = tokio::task::spawn_blocking(move || -> TestResult {
-    guest(Ftp::open("ftp".into(), config(port)))?;
-    let first = guest(Ftp::list(Locator::Path("/".into()), None, 10))?;
-    assert_eq!(first.entries[0].name, " report.md ");
-    wait_expired.blocking_recv()?;
-    let next = guest(Ftp::list(Locator::Path("/".into()), None, 10))?;
-    assert_eq!(next.entries[0].name, " report.md ");
-    Ftp::close();
-    Ok(())
-  })
-  .await?;
-  if client.is_err() {
-    server.abort();
-  }
-  client?;
-  server.await??;
   Ok(())
 }
 
 #[tokio::test]
 async fn recovery_before_a_mutation_never_replays_a_lost_mutation_reply()
 -> TestResult {
-  let listener = TcpListener::bind("127.0.0.1:0").await?;
+  let fixture = test_host::install(test_host::Scenario::LostMutationReply);
-  let port = listener.local_addr()?.port();
+  guest(Ftp::open("ftp".into(), config()))?;
-  let (expired, wait_expired) = oneshot::channel();
+  assert!(matches!(
-  let server = tokio::spawn(async move {
+    Ftp::mkdir(Locator::Path("/new".into())),
-    tokio::time::timeout(Duration::from_secs(5), async {
+    Err(ErrorCode::Unavailable)
-      let mut channel = login(&listener).await?;
+  ));
-      channel.get_mut().shutdown().await?;
+  Ftp::close();
-      drop(channel);
+  let summary = fixture.summary();
-      expired.send(()).map_err(|_| "guest disappeared")?;
+  assert_eq!(summary.control_connections, 2);
-      let mut channel = login(&listener).await?;
+  assert_eq!(summary.listings, 0);
-      expect(&mut channel, "MKD /new").await?;
+  assert_eq!(summary.mkdir_commands, 1);
-      // The mutation may have succeeded; lose its reply deliberately.
-      channel.get_mut().shutdown().await?;
-      drop(channel);
-      Ok::<_, Box<dyn std::error::Error + Send + Sync>>(listener)
-    })
-    .await?
-  });
-  // `Guest` is generated as a synchronous trait, so the repository rule for
-  // unavoidable blocking I/O requires isolating this native test call here.
-  let client = tokio::task::spawn_blocking(move || -> TestResult {
-    guest(Ftp::open("ftp".into(), config(port)))?;
-    wait_expired.blocking_recv()?;
-    assert!(matches!(
-      Ftp::mkdir(Locator::Path("/new".into())),
-      Err(ErrorCode::Unavailable)
-    ));
-    Ftp::close();
-    Ok(())
-  })
-  .await?;
-  if client.is_err() {
-    server.abort();
-  }
-  client?;
-  let listener = server.await??;
-  assert!(
-    tokio::time::timeout(Duration::from_millis(50), listener.accept())
-      .await
-      .is_err(),
-    "a mutation must not be replayed on a third control connection"
-  );
   Ok(())
 }

src/test_host.rs

@@ -1,51 +1,166 @@
-//! Native test transport for the generated synchronous `Guest` API. It cannot
+//! In-memory FTP host for the generated synchronous `Guest` API. Production
-//! await a Tokio socket, so tests confine it to `spawn_blocking` while their FTP
+//! WASM uses Memona's real async socket host; live QA covers that boundary.
-//! server remains async. Production WASM always uses the host's async sockets.
 
 use std::cell::RefCell;
-use std::io::{Read, Write};
+use std::collections::VecDeque;
-use std::net::{IpAddr, SocketAddr, TcpStream};
+use std::rc::Rc;
-use std::time::Duration;
 
 pub use memona_plugin_sdk::HostError;
 
-pub struct Tcp(RefCell<TcpStream>);
+pub const CONTROL_PORT: u16 = 21_210;
+const DATA_PORT: u16 = 31_310;
+const LISTING: &[u8] = b"type=file;size=3;  report.md \r\n";
+
+#[derive(Clone, Copy)]
+pub enum Scenario {
+  ExpiredListing,
+  LostMutationReply,
+}
+
+struct State {
+  scenario: Scenario,
+  control_connections: usize,
+  listings: usize,
+  mkdir_commands: usize,
+}
+
+thread_local! {
+  static ACTIVE: RefCell<Option<Rc<RefCell<State>>>> = const { RefCell::new(None) };
+}
+
+pub struct Fixture(Rc<RefCell<State>>);
+
+pub struct Summary {
+  pub control_connections: usize,
+  pub listings: usize,
+  pub mkdir_commands: usize,
+}
+
+pub fn install(scenario: Scenario) -> Fixture {
+  let state = Rc::new(RefCell::new(State {
+    scenario,
+    control_connections: 0,
+    listings: 0,
+    mkdir_commands: 0,
+  }));
+  ACTIVE.set(Some(state.clone()));
+  Fixture(state)
+}
+
+impl Fixture {
+  pub fn summary(&self) -> Summary {
+    let state = self.0.borrow();
+    Summary {
+      control_connections: state.control_connections,
+      listings: state.listings,
+      mkdir_commands: state.mkdir_commands,
+    }
+  }
+}
+
+impl Drop for Fixture {
+  fn drop(&mut self) {
+    ACTIVE.with_borrow_mut(|active| {
+      if active
+        .as_ref()
+        .is_some_and(|state| Rc::ptr_eq(state, &self.0))
+      {
+        *active = None;
+      }
+    });
+  }
+}
+
+enum Kind {
+  Control(usize),
+  Data,
+}
+
+pub struct Tcp {
+  state: Rc<RefCell<State>>,
+  kind: Kind,
+  input: RefCell<VecDeque<u8>>,
+}
 
 pub fn connect(name: &str, port: u16, tls: bool) -> Result<Tcp, HostError> {
-  if tls {
+  if tls || name != "127.0.0.1" {
     return Err(HostError::InvalidInput);
   }
-  let ip: IpAddr = name.parse().map_err(|_| HostError::InvalidInput)?;
+  let state = ACTIVE
-  let timeout = Duration::from_secs(2);
+    .with_borrow(Clone::clone)
-  let stream = TcpStream::connect_timeout(&SocketAddr::new(ip, port), timeout)
+    .ok_or(HostError::Internal)?;
-    .map_err(|_| HostError::Unavailable)?;
+  match port {
-  stream
+    CONTROL_PORT => {
-    .set_read_timeout(Some(timeout))
+      let index = {
-    .map_err(|_| HostError::Unavailable)?;
+        let mut state = state.borrow_mut();
-  stream
+        let index = state.control_connections;
-    .set_write_timeout(Some(timeout))
+        state.control_connections += 1;
-    .map_err(|_| HostError::Unavailable)?;
+        index
-  Ok(Tcp(RefCell::new(stream)))
+      };
+      Ok(Tcp {
+        state,
+        kind: Kind::Control(index),
+        input: RefCell::new(VecDeque::from(b"220 Fixture\r\n".to_vec())),
+      })
+    }
+    DATA_PORT => Ok(Tcp {
+      state,
+      kind: Kind::Data,
+      input: RefCell::new(VecDeque::from(LISTING.to_vec())),
+    }),
+    _ => Err(HostError::Denied),
+  }
 }
 
 impl Tcp {
   pub fn read(&self, limit: u32) -> Result<Vec<u8>, HostError> {
-    let mut bytes = vec![0; limit as usize];
+    let mut input = self.input.borrow_mut();
-    let count = self
+    if input.is_empty() && matches!(self.kind, Kind::Control(_)) {
-      .0
+      return Err(HostError::Unavailable);
-      .borrow_mut()
+    }
-      .read(&mut bytes)
+    let count = usize::min(limit as usize, input.len());
-      .map_err(|_| HostError::Unavailable)?;
+    Ok(input.drain(..count).collect())
-    bytes.truncate(count);
-    Ok(bytes)
   }
 
   pub fn write(&self, bytes: &[u8]) -> Result<u32, HostError> {
-    self
+    let Kind::Control(index) = self.kind else {
-      .0
+      return Err(HostError::InvalidInput);
-      .borrow_mut()
+    };
-      .write(bytes)
+    let command = std::str::from_utf8(bytes)
-      .map(|count| count as u32)
+      .map_err(|_| HostError::InvalidInput)?
-      .map_err(|_| HostError::Unavailable)
+      .strip_suffix("\r\n")
+      .ok_or(HostError::InvalidInput)?;
+    let response = {
+      let mut state = self.state.borrow_mut();
+      match command {
+        "USER fixture" => "331 Password required\r\n",
+        "PASS  fixture secret " => "230 Signed in\r\n",
+        "TYPE I" => "200 Binary\r\n",
+        "QUIT" => "221 Bye\r\n",
+        "EPSV" => "229 Passive (|||31310|)\r\n",
+        "MLSD /" if matches!(state.scenario, Scenario::ExpiredListing) => {
+          state.listings += 1;
+          "150 Listing\r\n226 Complete\r\n"
+        }
+        "NOOP"
+          if index == 0
+            && (matches!(state.scenario, Scenario::LostMutationReply)
+              || state.listings > 0) =>
+        {
+          return Err(HostError::Unavailable);
+        }
+        "NOOP" => "200 Still connected\r\n",
+        "MKD /new"
+          if index == 1
+            && matches!(state.scenario, Scenario::LostMutationReply) =>
+        {
+          state.mkdir_commands += 1;
+          return Ok(bytes.len() as u32);
+        }
+        _ => return Err(HostError::InvalidInput),
+      }
+    };
+    self.input.borrow_mut().extend(response.as_bytes());
+    Ok(bytes.len() as u32)
   }
 }