Report uncertain FTP mutation outcomes

src/control.rs

@@ -91,6 +91,11 @@ pub struct Control {
   buffer: Vec<u8>,
 }
 
+struct DispatchError {
+  code: ErrorCode,
+  dispatched: bool,
+}
+
 impl Control {
   pub fn open(
     host_name: String,
@@ -155,27 +160,61 @@ impl Control {
     Err(ErrorCode::Unavailable)
   }
 
-  /// Sends one command. A newline in an argument would forge a second command,
-  /// so it is refused rather than escaped.
-  pub fn send(&mut self, command: &str) -> Result<(u16, String), ErrorCode> {
+  /// Writes one complete command and records whether any of it reached the
+  /// socket. A newline in an argument would forge a second command, so it is
+  /// refused before dispatch rather than escaped.
+  fn dispatch(&mut self, command: &str) -> Result<(), DispatchError> {
     if command.contains(['\r', '\n']) || command.len() > MAX_LINE {
-      return Err(ErrorCode::InvalidInput);
+      return Err(DispatchError {
+        code: ErrorCode::InvalidInput,
+        dispatched: false,
+      });
     }
     let line = format!("{command}\r\n");
     let mut sent = 0;
     while sent < line.len() {
-      let written = self
+      let written =
+        self
           .socket
           .write(&line.as_bytes()[sent..])
-        .map_err(transport)? as usize;
+          .map_err(|error| DispatchError {
+            code: transport(error),
+            dispatched: sent > 0,
+          })? as usize;
       if written == 0 {
-        return Err(ErrorCode::Unavailable);
+        return Err(DispatchError {
+          code: ErrorCode::Unavailable,
+          dispatched: sent > 0,
+        });
       }
       sent += written;
     }
+    Ok(())
+  }
+
+  /// Sends one command and reads its reply.
+  pub fn send(&mut self, command: &str) -> Result<(u16, String), ErrorCode> {
+    self.dispatch(command).map_err(|error| error.code)?;
     self.reply()
   }
 
+  /// Sends a command that changes remote state. Once any command byte was
+  /// dispatched, a write interruption or missing reply makes its outcome
+  /// unknowable even when the underlying transport reports cancellation.
+  pub fn send_mutation(
+    &mut self,
+    command: &str,
+  ) -> Result<(u16, String), ErrorCode> {
+    self.dispatch(command).map_err(|error| {
+      if error.dispatched {
+        ErrorCode::OutcomeUnknown
+      } else {
+        error.code
+      }
+    })?;
+    self.reply().map_err(|_| ErrorCode::OutcomeUnknown)
+  }
+
   pub fn expect(&mut self, command: &str) -> Result<String, ErrorCode> {
     let (code, text) = self.send(command)?;
     outcome(code)?;

src/lib.rs

@@ -89,7 +89,14 @@ fn with_control<T>(
       Some(control) => action(control),
       None => Err(ErrorCode::Internal),
     };
-    if matches!(result, Err(ErrorCode::Unavailable | ErrorCode::Internal)) {
+    if matches!(
+      result,
+      Err(
+        ErrorCode::Unavailable
+          | ErrorCode::OutcomeUnknown
+          | ErrorCode::Internal
+      )
+    ) {
       *slot = None;
     }
     result
@@ -126,6 +133,17 @@ fn transfer(command: &str) -> Result<host::Tcp, ErrorCode> {
   })
 }
 
+/// Starts a passive upload. Losing the command reply is ambiguous because an
+/// FTP server may have created or truncated the destination on receipt.
+fn mutation_transfer(command: &str) -> Result<host::Tcp, ErrorCode> {
+  with_control(|control| {
+    let data = control.passive()?;
+    let (code, _) = control.send_mutation(command)?;
+    preliminary(code)?;
+    Ok(data)
+  })
+}
+
 /// Consumes the completion reply of a transfer whose data socket is closed.
 fn settle() -> Result<(), ErrorCode> {
   with_control(|control| {
@@ -134,6 +152,16 @@ fn settle() -> Result<(), ErrorCode> {
   })
 }
 
+/// Consumes an upload's completion reply. The data and command are already
+/// dispatched, so an interrupted reply cannot say whether the file committed.
+fn settle_mutation() -> Result<(), ErrorCode> {
+  let (code, _) = with_control(|control| {
+    control.reply().map_err(|_| ErrorCode::OutcomeUnknown)
+  })
+  .map_err(|_| ErrorCode::OutcomeUnknown)?;
+  outcome(code)
+}
+
 /// Ends a transfer early, after its data socket is closed.
 fn abort() -> Result<(), ErrorCode> {
   with_control(Control::abort)
@@ -237,9 +265,11 @@ impl GuestWriter for FtpWriter {
     let socket = socket.as_ref().ok_or(ErrorCode::InvalidInput)?;
     let mut sent = 0;
     while sent < data.len() {
-      let written = socket.write(&data[sent..]).map_err(transport)? as usize;
+      let written = socket
+        .write(&data[sent..])
+        .map_err(|_| ErrorCode::OutcomeUnknown)? as usize;
       if written == 0 {
-        return Err(ErrorCode::Unavailable);
+        return Err(ErrorCode::OutcomeUnknown);
       }
       sent += written;
     }
@@ -250,7 +280,7 @@ impl GuestWriter for FtpWriter {
   /// server answer, so the socket is given up before the reply is read.
   fn finish(&self) -> Result<Option<FileInfo>, ErrorCode> {
     drop(self.socket.take().ok_or(ErrorCode::InvalidInput)?);
-    settle()?;
+    settle_mutation()?;
     Ok(Some(FileInfo {
       kind: EntryKind::File,
       size: self.written.get(),
@@ -269,8 +299,11 @@ impl Drop for FtpWriter {
     };
     drop(socket);
     if abort().is_ok() {
-      let _ =
-        with_control(|control| control.expect(&format!("DELE {}", self.path)));
+      let _ = with_control(|control| {
+        let (code, _) =
+          control.send_mutation(&format!("DELE {}", self.path))?;
+        outcome(code)
+      });
     }
   }
 }
@@ -394,7 +427,7 @@ impl Guest for Ftp {
       return Err(ErrorCode::Unsupported);
     }
     ready()?;
-    let socket = transfer(&format!("STOR {path}"))?;
+    let socket = mutation_transfer(&format!("STOR {path}"))?;
     Ok(Writer::new(FtpWriter {
       path,
       socket: RefCell::new(Some(socket)),
@@ -408,7 +441,7 @@ impl Guest for Ftp {
     let path = route(&target)?;
     ready()?;
     with_control(|control| {
-      let (code, text) = control.send(&format!("MKD {path}"))?;
+      let (code, text) = control.send_mutation(&format!("MKD {path}"))?;
       refusal(code, &text)
     })?;
     Ok(FileInfo {
@@ -427,14 +460,16 @@ impl Guest for Ftp {
     let path = route(&target)?;
     ready()?;
     with_control(|control| {
+      let info = probe(control, &path)?;
       match probe(control, &destination) {
         Ok(_) => return Err(ErrorCode::Conflict),
         Err(ErrorCode::NotFound) => {}
         Err(error) => return Err(error),
       }
       control.expect(&format!("RNFR {path}"))?;
-      control.expect(&format!("RNTO {destination}"))?;
-      probe(control, &destination)
+      let (code, _) = control.send_mutation(&format!("RNTO {destination}"))?;
+      outcome(code)?;
+      Ok(info)
     })
   }
 
@@ -445,11 +480,13 @@ impl Guest for Ftp {
     let path = route(&target)?;
     ready()?;
     with_control(|control| {
-      let (code, _) = control.send(&format!("DELE {path}"))?;
-      if (200..300).contains(&code) {
-        return Ok(());
+      let (code, _) = control.send_mutation(&format!("DELE {path}"))?;
+      match outcome(code) {
+        Ok(()) => return Ok(()),
+        Err(ErrorCode::NotFound) => {}
+        Err(error) => return Err(error),
       }
-      let (code, text) = control.send(&format!("RMD {path}"))?;
+      let (code, text) = control.send_mutation(&format!("RMD {path}"))?;
       refusal(code, &text)
     })
   }

src/session_tests.rs

@@ -1,7 +1,7 @@
 //! FTP command recovery against an in-memory protocol transport. Live QA
 //! separately exercises the compiled WASM component over real host sockets.
 
-use super::{Ftp, test_host};
+use super::{Ftp, mutation_transfer, settle_mutation, test_host};
 use memona_plugin_sdk::{ErrorCode, Guest, Locator, Setting};
 
 type TestResult<T = ()> = Result<T, Box<dyn std::error::Error + Send + Sync>>;
@@ -49,7 +49,7 @@ async fn recovery_before_a_mutation_never_replays_a_lost_mutation_reply()
   guest(Ftp::open("ftp".into(), config()))?;
   assert!(matches!(
     Ftp::mkdir(Locator::Path("/new".into())),
-    Err(ErrorCode::Unavailable)
+    Err(ErrorCode::OutcomeUnknown)
   ));
   Ftp::close();
   let summary = fixture.summary();
@@ -75,3 +75,19 @@ async fn cancelled_preflight_never_reconnects_or_executes_the_mutation()
   assert_eq!(summary.mkdir_commands, 0);
   Ok(())
 }
+
+#[tokio::test]
+async fn a_lost_upload_completion_reports_an_unknown_outcome() -> TestResult {
+  let fixture = test_host::install(test_host::Scenario::LostUploadCompletion);
+  guest(Ftp::open("ftp".into(), config()))?;
+  let socket = guest(mutation_transfer("STOR /upload"))?;
+  assert!(matches!(socket.write(b"body"), Ok(4)));
+  drop(socket);
+  assert!(matches!(settle_mutation(), Err(ErrorCode::OutcomeUnknown)));
+  Ftp::close();
+  let summary = fixture.summary();
+  assert_eq!(summary.control_connections, 1);
+  assert_eq!(summary.upload_commands, 1);
+  assert_eq!(summary.uploaded_bytes, 4);
+  Ok(())
+}

src/test_host.rs

@@ -16,6 +16,7 @@ pub enum Scenario {
   CancelledPreflight,
   ExpiredListing,
   LostMutationReply,
+  LostUploadCompletion,
 }
 
 struct State {
@@ -23,6 +24,8 @@ struct State {
   control_connections: usize,
   listings: usize,
   mkdir_commands: usize,
+  upload_commands: usize,
+  uploaded_bytes: usize,
 }
 
 thread_local! {
@@ -35,6 +38,8 @@ pub struct Summary {
   pub control_connections: usize,
   pub listings: usize,
   pub mkdir_commands: usize,
+  pub upload_commands: usize,
+  pub uploaded_bytes: usize,
 }
 
 pub fn install(scenario: Scenario) -> Fixture {
@@ -43,6 +48,8 @@ pub fn install(scenario: Scenario) -> Fixture {
     control_connections: 0,
     listings: 0,
     mkdir_commands: 0,
+    upload_commands: 0,
+    uploaded_bytes: 0,
   }));
   ACTIVE.set(Some(state.clone()));
   Fixture(state)
@@ -55,6 +62,8 @@ impl Fixture {
       control_connections: state.control_connections,
       listings: state.listings,
       mkdir_commands: state.mkdir_commands,
+      upload_commands: state.upload_commands,
+      uploaded_bytes: state.uploaded_bytes,
     }
   }
 }
@@ -124,8 +133,16 @@ impl Tcp {
   }
 
   pub fn write(&self, bytes: &[u8]) -> Result<u32, HostError> {
-    let Kind::Control(index) = self.kind else {
+    if matches!(self.kind, Kind::Data) {
+      let mut state = self.state.borrow_mut();
+      if matches!(state.scenario, Scenario::LostUploadCompletion) {
+        state.uploaded_bytes += bytes.len();
+        return Ok(bytes.len() as u32);
+      }
       return Err(HostError::InvalidInput);
+    }
+    let Kind::Control(index) = self.kind else {
+      unreachable!();
     };
     let command = std::str::from_utf8(bytes)
       .map_err(|_| HostError::InvalidInput)?
@@ -154,6 +171,12 @@ impl Tcp {
           return Err(HostError::Unavailable);
         }
         "NOOP" => "200 Still connected\r\n",
+        "STOR /upload"
+          if matches!(state.scenario, Scenario::LostUploadCompletion) =>
+        {
+          state.upload_commands += 1;
+          "150 Upload started\r\n"
+        }
         "MKD /new"
           if index == 1
             && matches!(state.scenario, Scenario::LostMutationReply) =>