Recover expired FTP control sessions

src/control.rs

@@ -1,6 +1,9 @@
 //! The FTP control channel: one socket, line replies, and passive transfers.
 
+#[cfg(test)]
+use crate::test_host as host;
 use memona_plugin_sdk::ErrorCode;
+#[cfg(not(test))]
 use memona_plugin_sdk::memona::filesystem::host;
 
 const MAX_LINE: usize = 8192;

src/lib.rs

@@ -4,8 +4,14 @@ mod control;
 mod listing;
 mod transfer;
 
+#[cfg(test)]
+mod session_tests;
+#[cfg(test)]
+mod test_host;
+
 use control::{Control, outcome, preliminary, refusal, transport};
 use listing::{machine_info, page};
+#[cfg(not(test))]
 use memona_plugin_sdk::memona::filesystem::host;
 use memona_plugin_sdk::{
   Capabilities, ChangePage, EntryKind, ErrorCode, FileInfo, Guest, GuestReader,
@@ -14,6 +20,8 @@ use memona_plugin_sdk::{
 use std::cell::{Cell, RefCell};
 use std::collections::HashMap;
 use std::rc::Rc;
+#[cfg(test)]
+use test_host as host;
 use transfer::Buffered;
 
 const MAX_CHUNK: u32 = 1024 * 1024;
@@ -22,6 +30,7 @@ const MAX_LISTING_BYTES: usize = 8 * 1024 * 1024;
 
 thread_local! {
   static CONTROL: RefCell<Option<Control>> = const { RefCell::new(None) };
+  static SESSION: RefCell<Option<Session>> = const { RefCell::new(None) };
   /// Raw MLSD listings by path, so later pages of a folder are parsed from
   /// memory instead of downloaded and parsed whole again.
   static LISTINGS: RefCell<HashMap<String, Rc<[u8]>>> = RefCell::new(HashMap::new());
@@ -29,6 +38,31 @@ thread_local! {
 
 struct Ftp;
 
+#[derive(Clone)]
+struct Session {
+  host: String,
+  port: u16,
+  tls: bool,
+  user: String,
+  password: String,
+}
+
+impl Session {
+  fn connect(&self) -> Result<Control, ErrorCode> {
+    let mut control = Control::open(self.host.clone(), self.port, self.tls)?;
+    let (code, _) = control.send(&format!("USER {}", self.user))?;
+    if code == 331 {
+      let (code, _) = control.send(&format!("PASS {}", self.password))?;
+      outcome(code)?;
+    } else {
+      outcome(code)?;
+    }
+    // Binary throughout: text mode would rewrite line endings inside files.
+    control.expect("TYPE I")?;
+    Ok(control)
+  }
+}
+
 fn text(setting: Option<&Setting>) -> Option<String> {
   match setting {
     Some(Setting::Text(value)) => {
@@ -50,12 +84,32 @@ fn route(target: &Locator) -> Result<String, ErrorCode> {
 fn with_control<T>(
   action: impl FnOnce(&mut Control) -> Result<T, ErrorCode>,
 ) -> Result<T, ErrorCode> {
-  CONTROL.with_borrow_mut(|slot| match slot.as_mut() {
-    Some(control) => action(control),
-    None => Err(ErrorCode::Internal),
+  CONTROL.with_borrow_mut(|slot| {
+    let result = match slot.as_mut() {
+      Some(control) => action(control),
+      None => Err(ErrorCode::Internal),
+    };
+    if matches!(result, Err(ErrorCode::Unavailable | ErrorCode::Internal)) {
+      *slot = None;
+    }
+    result
   })
 }
 
+/// Checks a control channel only before a fresh filesystem operation. A lost
+/// idle session is replaced and logged in again. Nothing inside a transfer or
+/// mutation retries, because its missing reply may hide a completed change.
+fn ready() -> Result<(), ErrorCode> {
+  if with_control(|control| control.expect("NOOP")).is_ok() {
+    return Ok(());
+  }
+  let session = SESSION
+    .with_borrow(Clone::clone)
+    .ok_or(ErrorCode::Internal)?;
+  CONTROL.set(Some(session.connect()?));
+  Ok(())
+}
+
 /// Starts a passive transfer and leaves the data socket open for the caller,
 /// who owes the server one more reply once that socket is closed: either the
 /// completion, read by `settle`, or the answer to an `abort`.
@@ -241,19 +295,16 @@ impl Guest for Ftp {
       config.iter().find(|(key, _)| key == "tls").map(|(_, v)| v),
       Some(Setting::Toggle(true))
     );
-    let mut control = Control::open(host_name, port, tls)?;
     let user = lookup("username").unwrap_or_else(|| "anonymous".to_owned());
-    let (code, _) = control.send(&format!("USER {user}"))?;
-    if code == 331 {
-      let password = lookup("password").unwrap_or_default();
-      let (code, _) = control.send(&format!("PASS {password}"))?;
-      outcome(code)?;
-    } else {
-      outcome(code)?;
-    }
-    // Binary throughout: text mode would rewrite line endings inside files.
-    control.expect("TYPE I")?;
-    CONTROL.set(Some(control));
+    let session = Session {
+      host: host_name,
+      port,
+      tls,
+      user,
+      password: lookup("password").unwrap_or_default(),
+    };
+    CONTROL.set(Some(session.connect()?));
+    SESSION.set(Some(session));
     Ok(Capabilities {
       write: true,
       mkdir: true,
@@ -274,6 +325,7 @@ impl Guest for Ftp {
       }
       *slot = None;
     });
+    SESSION.set(None);
   }
 
   fn stat(target: Locator) -> Result<FileInfo, ErrorCode> {
@@ -285,6 +337,7 @@ impl Guest for Ftp {
         revision: None,
       });
     }
+    ready()?;
     with_control(|control| probe(control, &path))
   }
 
@@ -306,13 +359,17 @@ impl Guest for Ftp {
     });
     let listing = match kept {
       Some(listing) => listing,
-      None => fetch_listing(&path)?,
+      None => {
+        ready()?;
+        fetch_listing(&path)?
+      }
     };
     page(&listing, start, limit as usize)
   }
 
   fn open_read(target: Locator, offset: u64) -> Result<Reader, ErrorCode> {
     let path = route(&target)?;
+    ready()?;
     if offset > 0 {
       with_control(|control| control.expect(&format!("REST {offset}")))?;
     }
@@ -332,6 +389,7 @@ impl Guest for Ftp {
     if !matches!(condition, WriteCondition::Overwrite) {
       return Err(ErrorCode::Unsupported);
     }
+    ready()?;
     let socket = transfer(&format!("STOR {path}"))?;
     Ok(Writer::new(FtpWriter {
       path,
@@ -344,6 +402,7 @@ impl Guest for Ftp {
   /// the one thing the server says in words rather than in its reply code.
   fn mkdir(target: Locator) -> Result<FileInfo, ErrorCode> {
     let path = route(&target)?;
+    ready()?;
     with_control(|control| {
       let (code, text) = control.send(&format!("MKD {path}"))?;
       refusal(code, &text)
@@ -362,6 +421,7 @@ impl Guest for Ftp {
     destination: String,
   ) -> Result<FileInfo, ErrorCode> {
     let path = route(&target)?;
+    ready()?;
     with_control(|control| {
       match probe(control, &destination) {
         Ok(_) => return Err(ErrorCode::Conflict),
@@ -379,6 +439,7 @@ impl Guest for Ftp {
   /// holds children is refused as a conflict, not as an absence.
   fn remove(target: Locator) -> Result<(), ErrorCode> {
     let path = route(&target)?;
+    ready()?;
     with_control(|control| {
       let (code, _) = control.send(&format!("DELE {path}"))?;
       if (200..300).contains(&code) {
@@ -394,6 +455,7 @@ impl Guest for Ftp {
   }
 }
 
+#[cfg(not(test))]
 memona_plugin_sdk::export!(Ftp);
 
 #[cfg(test)]

src/session_tests.rs

@@ -0,0 +1,162 @@
+//! Real FTP commands over a native test socket, with an async local server.
+//! This verifies guest recovery; WASM host integration is exercised separately.
+
+use super::Ftp;
+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)> {
+  vec![
+    ("host".into(), Setting::Text("127.0.0.1".into())),
+    ("port".into(), Setting::Text(port.to_string())),
+    ("username".into(), Setting::Text("fixture".into())),
+    (
+      "password".into(),
+      Setting::Secret(" fixture secret ".into()),
+    ),
+  ]
+}
+
+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 port = listener.local_addr()?.port();
+  let (expired, wait_expired) = oneshot::channel();
+  let server = tokio::spawn(async move {
+    tokio::time::timeout(Duration::from_secs(5), async {
+      let mut channel = login(&listener).await?;
+      listing(&mut channel).await?;
+      reply(&mut channel, "421 Control connection timed out\r\n").await?;
+      channel.get_mut().shutdown().await?;
+      drop(channel);
+      expired.send(()).map_err(|_| "guest disappeared")?;
+      let mut channel = login(&listener).await?;
+      listing(&mut channel).await?;
+      expect(&mut channel, "QUIT").await?;
+      reply(&mut channel, "221 Bye\r\n").await
+    })
+    .await?
+  });
+  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 port = listener.local_addr()?.port();
+  let (expired, wait_expired) = oneshot::channel();
+  let server = tokio::spawn(async move {
+    tokio::time::timeout(Duration::from_secs(5), async {
+      let mut channel = login(&listener).await?;
+      channel.get_mut().shutdown().await?;
+      drop(channel);
+      expired.send(()).map_err(|_| "guest disappeared")?;
+      let mut channel = login(&listener).await?;
+      expect(&mut channel, "MKD /new").await?;
+      // 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?
+  });
+  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

@@ -0,0 +1,50 @@
+//! Native test transport. Guest calls run only inside `spawn_blocking`;
+//! production WASM always uses the host's sockets and network policy.
+
+use std::cell::RefCell;
+use std::io::{Read, Write};
+use std::net::{IpAddr, SocketAddr, TcpStream};
+use std::time::Duration;
+
+pub use memona_plugin_sdk::HostError;
+
+pub struct Tcp(RefCell<TcpStream>);
+
+pub fn connect(name: &str, port: u16, tls: bool) -> Result<Tcp, HostError> {
+  if tls {
+    return Err(HostError::InvalidInput);
+  }
+  let ip: IpAddr = name.parse().map_err(|_| HostError::InvalidInput)?;
+  let timeout = Duration::from_secs(2);
+  let stream = TcpStream::connect_timeout(&SocketAddr::new(ip, port), timeout)
+    .map_err(|_| HostError::Unavailable)?;
+  stream
+    .set_read_timeout(Some(timeout))
+    .map_err(|_| HostError::Unavailable)?;
+  stream
+    .set_write_timeout(Some(timeout))
+    .map_err(|_| HostError::Unavailable)?;
+  Ok(Tcp(RefCell::new(stream)))
+}
+
+impl Tcp {
+  pub fn read(&self, limit: u32) -> Result<Vec<u8>, HostError> {
+    let mut bytes = vec![0; limit as usize];
+    let count = self
+      .0
+      .borrow_mut()
+      .read(&mut bytes)
+      .map_err(|_| HostError::Unavailable)?;
+    bytes.truncate(count);
+    Ok(bytes)
+  }
+
+  pub fn write(&self, bytes: &[u8]) -> Result<u32, HostError> {
+    self
+      .0
+      .borrow_mut()
+      .write(bytes)
+      .map(|count| count as u32)
+      .map_err(|_| HostError::Unavailable)
+  }
+}