From 70b0cf146bb0f8e00e7a4f024a31b6d1c278874d Mon Sep 17 00:00:00 2001 From: Danny Kim Date: Wed, 30 Sep 2026 20:22:22 +0900 Subject: [PATCH] Report uncertain FTP mutation outcomes --- src/control.rs | 57 ++++++++++++++++++++++++++++++++------ src/lib.rs | 65 ++++++++++++++++++++++++++++++++++---------- src/session_tests.rs | 20 ++++++++++++-- src/test_host.rs | 25 ++++++++++++++++- 4 files changed, 141 insertions(+), 26 deletions(-) diff --git a/src/control.rs b/src/control.rs index fb0c3b8..1cf2022 100644 --- a/src/control.rs +++ b/src/control.rs @@ -91,6 +91,11 @@ pub struct Control { buffer: Vec, } +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 - .socket - .write(&line.as_bytes()[sent..]) - .map_err(transport)? as usize; + let written = + self + .socket + .write(&line.as_bytes()[sent..]) + .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 { let (code, text) = self.send(command)?; outcome(code)?; diff --git a/src/lib.rs b/src/lib.rs index cbc279b..4cb2613 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -89,7 +89,14 @@ fn with_control( 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 { }) } +/// 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 { + 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, 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) }) } diff --git a/src/session_tests.rs b/src/session_tests.rs index 8e059fa..611b1a1 100644 --- a/src/session_tests.rs +++ b/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 = Result>; @@ -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(()) +} diff --git a/src/test_host.rs b/src/test_host.rs index cff73ec..98a1bf7 100644 --- a/src/test_host.rs +++ b/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 { - 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) =>