Stage FTP uploads before promotion

src/lib.rs

@@ -27,10 +27,12 @@ use transfer::Buffered;
 const MAX_CHUNK: u32 = 1024 * 1024;
 /// One listing may be this large, and so may every listing kept together.
 const MAX_LISTING_BYTES: usize = 8 * 1024 * 1024;
+const MAX_STAGE_ATTEMPTS: usize = 64;
 
 thread_local! {
   static CONTROL: RefCell<Option<Control>> = const { RefCell::new(None) };
   static SESSION: RefCell<Option<Session>> = const { RefCell::new(None) };
+  static NEXT_STAGE: Cell<u64> = const { Cell::new(0) };
   /// 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());
@@ -254,18 +256,38 @@ impl Drop for FtpReader {
 
 /// Writes one file to a passive data socket.
 struct FtpWriter {
-  path: String,
+  target: String,
+  staged: RefCell<Option<String>>,
   socket: RefCell<Option<host::Tcp>>,
   written: Cell<u64>,
 }
 
 impl FtpWriter {
+  /// Removes the private staging file when possible. A missing stage means a
+  /// prior promotion or cleanup already completed.
+  fn cleanup_stage(&self) {
+    let Some(staged) = self.staged.borrow().clone() else {
+      return;
+    };
+    if ready().is_err() {
+      return;
+    }
+    let removed = with_control(|control| {
+      let (code, _) = control.send_mutation(&format!("DELE {staged}"))?;
+      outcome(code)
+    });
+    if matches!(removed, Ok(()) | Err(ErrorCode::NotFound)) {
+      self.staged.take();
+    }
+  }
+
   /// 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.
+  /// completion reply in an unknown state. Close both channels, then make a
+  /// best-effort cleanup of the private stage over a fresh session.
   fn discard_uncertain_upload(&self) -> ErrorCode {
     drop(self.socket.take());
     CONTROL.set(None);
+    self.cleanup_stage();
     ErrorCode::OutcomeUnknown
   }
 }
@@ -297,10 +319,30 @@ impl GuestWriter for FtpWriter {
     Ok(())
   }
   /// Closing the data socket is what ends an FTP upload; only then does the
-  /// server answer, so the socket is given up before the reply is read.
+  /// server answer. A confirmed stage is promoted only after that reply, so an
+  /// interrupted transfer never exposes a partial destination.
   fn finish(&self) -> Result<Option<FileInfo>, ErrorCode> {
     drop(self.socket.take().ok_or(ErrorCode::InvalidInput)?);
-    settle_mutation()?;
+    if let Err(error) = settle_mutation() {
+      self.cleanup_stage();
+      return Err(error);
+    }
+    let staged = self
+      .staged
+      .borrow()
+      .clone()
+      .ok_or(ErrorCode::InvalidInput)?;
+    let promoted = with_control(|control| {
+      control.expect(&format!("RNFR {staged}"))?;
+      let (code, _) =
+        control.send_mutation(&format!("RNTO {}", self.target))?;
+      outcome(code)
+    });
+    if let Err(error) = promoted {
+      self.cleanup_stage();
+      return Err(error);
+    }
+    self.staged.take();
     Ok(Some(FileInfo {
       kind: EntryKind::File,
       size: self.written.get(),
@@ -310,22 +352,51 @@ impl GuestWriter for FtpWriter {
 }
 
 impl Drop for FtpWriter {
-  /// An upload abandoned before it finished must leave no partial file. The
-  /// server takes the closed socket as the end of the upload, so what it
-  /// stored is deleted once the channel is back in sync.
+  /// An abandoned upload closes its data channel, resynchronizes when
+  /// possible, and removes only its private stage. The destination was never
+  /// exposed unless `finish` confirmed the promotion.
   fn drop(&mut self) {
-    let Some(socket) = self.socket.take() else {
-      return;
-    };
-    drop(socket);
-    if abort().is_ok() {
-      let _ = with_control(|control| {
-        let (code, _) =
-          control.send_mutation(&format!("DELE {}", self.path))?;
-        outcome(code)
-      });
+    if let Some(socket) = self.socket.take() {
+      drop(socket);
+      if abort().is_err() {
+        CONTROL.set(None);
+      }
+    }
+    self.cleanup_stage();
+  }
+}
+
+/// Chooses an unused private sibling for one upload. The sequence is local to
+/// the component, and the remote probe prevents reuse after an app restart.
+fn stage_path(target: &str) -> Result<String, ErrorCode> {
+  let (parent, name) =
+    target.rsplit_once('/').ok_or(ErrorCode::InvalidInput)?;
+  if name.is_empty() {
+    return Err(ErrorCode::InvalidInput);
+  }
+  for _ in 0..MAX_STAGE_ATTEMPTS {
+    let sequence = NEXT_STAGE.get();
+    NEXT_STAGE.set(sequence.wrapping_add(1));
+    let staged = format!("{parent}/.memona-upload-{sequence:016x}");
+    match with_control(|control| probe(control, &staged)) {
+      Err(ErrorCode::NotFound) => return Ok(staged),
+      Ok(_) => {}
+      Err(error) => return Err(error),
     }
   }
+  Err(ErrorCode::Conflict)
+}
+
+fn start_writer(target: String) -> Result<FtpWriter, ErrorCode> {
+  ready()?;
+  let staged = stage_path(&target)?;
+  let socket = mutation_transfer(&format!("STOR {staged}"))?;
+  Ok(FtpWriter {
+    target,
+    staged: RefCell::new(Some(staged)),
+    socket: RefCell::new(Some(socket)),
+    written: Cell::new(0),
+  })
 }
 
 impl Guest for Ftp {
@@ -446,13 +517,7 @@ impl Guest for Ftp {
     if !matches!(condition, WriteCondition::Overwrite) {
       return Err(ErrorCode::Unsupported);
     }
-    ready()?;
-    let socket = mutation_transfer(&format!("STOR {path}"))?;
-    Ok(Writer::new(FtpWriter {
-      path,
-      socket: RefCell::new(Some(socket)),
-      written: Cell::new(0),
-    }))
+    Ok(Writer::new(start_writer(path)?))
   }
 
   /// A refused create is a conflict when the path is already taken, which is

src/session_tests.rs

@@ -1,9 +1,8 @@
 //! FTP command recovery against an in-memory protocol transport. Live QA
 //! separately exercises the compiled WASM component over real host sockets.
 
-use super::{Ftp, FtpWriter, mutation_transfer, settle_mutation, test_host};
+use super::{Ftp, start_writer, 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>>;
 
@@ -81,15 +80,18 @@ async fn cancelled_preflight_never_reconnects_or_executes_the_mutation()
 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)));
+  let writer = guest(start_writer("/target".into()))?;
+  guest(writer.push(b"body".to_vec()))?;
+  assert!(matches!(writer.finish(), Err(ErrorCode::OutcomeUnknown)));
+  drop(writer);
   Ftp::close();
   let summary = fixture.summary();
-  assert_eq!(summary.control_connections, 1);
+  assert_eq!(summary.control_connections, 2);
   assert_eq!(summary.upload_commands, 1);
+  assert_eq!(summary.target_upload_commands, 0);
   assert_eq!(summary.uploaded_bytes, 4);
+  assert_eq!(summary.stage_deletes, 1);
+  assert_eq!(summary.promotions, 0);
   Ok(())
 }
 
@@ -98,11 +100,7 @@ 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),
-  };
+  let writer = guest(start_writer("/target".into()))?;
   assert!(matches!(
     writer.push(b"body".to_vec()),
     Err(ErrorCode::OutcomeUnknown)
@@ -114,7 +112,48 @@ async fn an_uncertain_upload_write_discards_the_control_session() -> TestResult
   let summary = fixture.summary();
   assert_eq!(summary.control_connections, 2);
   assert_eq!(summary.upload_commands, 1);
+  assert_eq!(summary.target_upload_commands, 0);
   assert_eq!(summary.uploaded_bytes, 0);
+  assert_eq!(summary.stage_deletes, 1);
+  assert_eq!(summary.promotions, 0);
   assert_eq!(summary.listings, 1);
   Ok(())
 }
+
+#[tokio::test]
+async fn a_complete_upload_promotes_its_private_stage() -> TestResult {
+  let fixture = test_host::install(test_host::Scenario::SuccessfulUpload);
+  guest(Ftp::open("ftp".into(), config()))?;
+  let writer = guest(start_writer("/target".into()))?;
+  guest(writer.push(b"body".to_vec()))?;
+  let info = guest(writer.finish())?.ok_or("missing upload result")?;
+  assert_eq!(info.size, 4);
+  drop(writer);
+  Ftp::close();
+  let summary = fixture.summary();
+  assert_eq!(summary.control_connections, 1);
+  assert_eq!(summary.upload_commands, 1);
+  assert_eq!(summary.target_upload_commands, 0);
+  assert_eq!(summary.uploaded_bytes, 4);
+  assert_eq!(summary.stage_deletes, 0);
+  assert_eq!(summary.promotions, 1);
+  Ok(())
+}
+
+#[tokio::test]
+async fn dropping_an_upload_removes_only_its_private_stage() -> TestResult {
+  let fixture = test_host::install(test_host::Scenario::AbandonedUpload);
+  guest(Ftp::open("ftp".into(), config()))?;
+  let writer = guest(start_writer("/target".into()))?;
+  guest(writer.push(b"partial".to_vec()))?;
+  drop(writer);
+  Ftp::close();
+  let summary = fixture.summary();
+  assert_eq!(summary.control_connections, 1);
+  assert_eq!(summary.upload_commands, 1);
+  assert_eq!(summary.target_upload_commands, 0);
+  assert_eq!(summary.uploaded_bytes, 7);
+  assert_eq!(summary.stage_deletes, 1);
+  assert_eq!(summary.promotions, 0);
+  Ok(())
+}

src/test_host.rs

@@ -13,11 +13,13 @@ const LISTING: &[u8] = b"type=file;size=3;  report.md \r\n";
 
 #[derive(Clone, Copy)]
 pub enum Scenario {
+  AbandonedUpload,
   CancelledPreflight,
   ExpiredListing,
   LostUploadData,
   LostMutationReply,
   LostUploadCompletion,
+  SuccessfulUpload,
 }
 
 struct State {
@@ -26,7 +28,10 @@ struct State {
   listings: usize,
   mkdir_commands: usize,
   upload_commands: usize,
+  target_upload_commands: usize,
   uploaded_bytes: usize,
+  stage_deletes: usize,
+  promotions: usize,
 }
 
 thread_local! {
@@ -40,7 +45,10 @@ pub struct Summary {
   pub listings: usize,
   pub mkdir_commands: usize,
   pub upload_commands: usize,
+  pub target_upload_commands: usize,
   pub uploaded_bytes: usize,
+  pub stage_deletes: usize,
+  pub promotions: usize,
 }
 
 pub fn install(scenario: Scenario) -> Fixture {
@@ -50,7 +58,10 @@ pub fn install(scenario: Scenario) -> Fixture {
     listings: 0,
     mkdir_commands: 0,
     upload_commands: 0,
+    target_upload_commands: 0,
     uploaded_bytes: 0,
+    stage_deletes: 0,
+    promotions: 0,
   }));
   ACTIVE.set(Some(state.clone()));
   Fixture(state)
@@ -64,7 +75,10 @@ impl Fixture {
       listings: state.listings,
       mkdir_commands: state.mkdir_commands,
       upload_commands: state.upload_commands,
+      target_upload_commands: state.target_upload_commands,
       uploaded_bytes: state.uploaded_bytes,
+      stage_deletes: state.stage_deletes,
+      promotions: state.promotions,
     }
   }
 }
@@ -136,7 +150,12 @@ impl Tcp {
   pub fn write(&self, bytes: &[u8]) -> Result<u32, HostError> {
     if matches!(self.kind, Kind::Data) {
       let mut state = self.state.borrow_mut();
-      if matches!(state.scenario, Scenario::LostUploadCompletion) {
+      if matches!(
+        state.scenario,
+        Scenario::AbandonedUpload
+          | Scenario::LostUploadCompletion
+          | Scenario::SuccessfulUpload
+      ) {
         state.uploaded_bytes += bytes.len();
         return Ok(bytes.len() as u32);
       }
@@ -177,18 +196,53 @@ impl Tcp {
           return Err(HostError::Unavailable);
         }
         "NOOP" => "200 Still connected\r\n",
-        "STOR /upload"
-          if matches!(state.scenario, Scenario::LostUploadCompletion) =>
+        command if command.starts_with("MLST /.memona-upload-") => {
+          "550 Missing\r\n"
+        }
+        command
+          if command.starts_with("STOR /.memona-upload-")
+            && matches!(
+              state.scenario,
+              Scenario::AbandonedUpload | Scenario::LostUploadCompletion
+            ) =>
         {
           state.upload_commands += 1;
           "150 Upload started\r\n"
         }
-        "STOR /broken"
-          if matches!(state.scenario, Scenario::LostUploadData) =>
+        command
+          if command.starts_with("STOR /.memona-upload-")
+            && matches!(state.scenario, Scenario::LostUploadData) =>
         {
           state.upload_commands += 1;
           "150 Upload started\r\n"
         }
+        command
+          if command.starts_with("STOR /.memona-upload-")
+            && matches!(state.scenario, Scenario::SuccessfulUpload) =>
+        {
+          state.upload_commands += 1;
+          "150 Upload started\r\n226 Upload complete\r\n"
+        }
+        command if command.starts_with("STOR ") => {
+          state.target_upload_commands += 1;
+          "150 Target upload started\r\n"
+        }
+        command if command.starts_with("DELE /.memona-upload-") => {
+          state.stage_deletes += 1;
+          "250 Stage deleted\r\n"
+        }
+        command if command.starts_with("RNFR /.memona-upload-") => {
+          "350 Ready for destination\r\n"
+        }
+        "RNTO /target"
+          if matches!(state.scenario, Scenario::SuccessfulUpload) =>
+        {
+          state.promotions += 1;
+          "250 Promoted\r\n"
+        }
+        "ABOR" if matches!(state.scenario, Scenario::AbandonedUpload) => {
+          "426 Upload abandoned\r\n226 Abort complete\r\n"
+        }
         "MKD /new"
           if index == 1
             && matches!(state.scenario, Scenario::LostMutationReply) =>