diff --git a/src/lib.rs b/src/lib.rs index 39d386a..e02b80c 100644 --- a/src/lib.rs +++ b/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> = const { RefCell::new(None) }; static SESSION: RefCell> = const { RefCell::new(None) }; + static NEXT_STAGE: Cell = 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>> = 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>, socket: RefCell>, written: Cell, } 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, 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 { + 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 { + 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 diff --git a/src/session_tests.rs b/src/session_tests.rs index d0a3f25..a594aa9 100644 --- a/src/session_tests.rs +++ b/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 = Result>; @@ -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(()) +} diff --git a/src/test_host.rs b/src/test_host.rs index 6a22113..a83444a 100644 --- a/src/test_host.rs +++ b/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 { 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) =>