src/lib.rs

@@ -259,17 +259,37 @@ struct FtpWriter {
   written: Cell<u64>,
 }
 
+impl FtpWriter {
+  /// A failed data write can leave both the remote file and its pending
+  /// completion reply in an unknown state. Close both channels so the next
+  /// filesystem operation must start a fresh session.
+  fn discard_uncertain_upload(&self) -> ErrorCode {
+    drop(self.socket.take());
+    CONTROL.set(None);
+    ErrorCode::OutcomeUnknown
+  }
+}
+
 impl GuestWriter for FtpWriter {
   fn push(&self, data: Vec<u8>) -> Result<(), ErrorCode> {
-    let socket = self.socket.borrow();
-    let socket = socket.as_ref().ok_or(ErrorCode::InvalidInput)?;
+    if self.socket.borrow().is_none() {
+      return Err(ErrorCode::InvalidInput);
+    }
     let mut sent = 0;
     while sent < data.len() {
-      let written = socket
-        .write(&data[sent..])
-        .map_err(|_| ErrorCode::OutcomeUnknown)? as usize;
+      let result = {
+        let socket = self.socket.borrow();
+        let Some(socket) = socket.as_ref() else {
+          return Err(ErrorCode::InvalidInput);
+        };
+        socket.write(&data[sent..])
+      };
+      let written = match result {
+        Ok(written) => written as usize,
+        Err(_) => return Err(self.discard_uncertain_upload()),
+      };
       if written == 0 {
-        return Err(ErrorCode::OutcomeUnknown);
+        return Err(self.discard_uncertain_upload());
       }
       sent += written;
     }

src/session_tests.rs

@@ -1,8 +1,9 @@
 //! FTP command recovery against an in-memory protocol transport. Live QA
 //! separately exercises the compiled WASM component over real host sockets.
 
-use super::{Ftp, mutation_transfer, settle_mutation, test_host};
-use memona_plugin_sdk::{ErrorCode, Guest, Locator, Setting};
+use super::{Ftp, FtpWriter, mutation_transfer, settle_mutation, test_host};
+use memona_plugin_sdk::{ErrorCode, Guest, GuestWriter, Locator, Setting};
+use std::cell::{Cell, RefCell};
 
 type TestResult<T = ()> = Result<T, Box<dyn std::error::Error + Send + Sync>>;
 
@@ -91,3 +92,29 @@ async fn a_lost_upload_completion_reports_an_unknown_outcome() -> TestResult {
   assert_eq!(summary.uploaded_bytes, 4);
   Ok(())
 }
+
+#[tokio::test]
+async fn an_uncertain_upload_write_discards_the_control_session() -> TestResult
+{
+  let fixture = test_host::install(test_host::Scenario::LostUploadData);
+  guest(Ftp::open("ftp".into(), config()))?;
+  let writer = FtpWriter {
+    path: "/broken".into(),
+    socket: RefCell::new(Some(guest(mutation_transfer("STOR /broken"))?)),
+    written: Cell::new(0),
+  };
+  assert!(matches!(
+    writer.push(b"body".to_vec()),
+    Err(ErrorCode::OutcomeUnknown)
+  ));
+  let page = guest(Ftp::list(Locator::Path("/".into()), None, 10))?;
+  assert_eq!(page.entries[0].name, " report.md ");
+  drop(writer);
+  Ftp::close();
+  let summary = fixture.summary();
+  assert_eq!(summary.control_connections, 2);
+  assert_eq!(summary.upload_commands, 1);
+  assert_eq!(summary.uploaded_bytes, 0);
+  assert_eq!(summary.listings, 1);
+  Ok(())
+}

src/test_host.rs

@@ -15,6 +15,7 @@ const LISTING: &[u8] = b"type=file;size=3;  report.md \r\n";
 pub enum Scenario {
   CancelledPreflight,
   ExpiredListing,
+  LostUploadData,
   LostMutationReply,
   LostUploadCompletion,
 }
@@ -156,7 +157,12 @@ impl Tcp {
         "TYPE I" => "200 Binary\r\n",
         "QUIT" => "221 Bye\r\n",
         "EPSV" => "229 Passive (|||31310|)\r\n",
-        "MLSD /" if matches!(state.scenario, Scenario::ExpiredListing) => {
+        "MLSD /"
+          if matches!(
+            state.scenario,
+            Scenario::ExpiredListing | Scenario::LostUploadData
+          ) =>
+        {
           state.listings += 1;
           "150 Listing\r\n226 Complete\r\n"
         }
@@ -177,6 +183,12 @@ impl Tcp {
           state.upload_commands += 1;
           "150 Upload started\r\n"
         }
+        "STOR /broken"
+          if matches!(state.scenario, Scenario::LostUploadData) =>
+        {
+          state.upload_commands += 1;
+          "150 Upload started\r\n"
+        }
         "MKD /new"
           if index == 1
             && matches!(state.scenario, Scenario::LostMutationReply) =>