diff --git a/integration_tests/src/lib.rs b/integration_tests/src/lib.rs index 7f29b0e2c..310d45e1a 100644 --- a/integration_tests/src/lib.rs +++ b/integration_tests/src/lib.rs @@ -487,6 +487,17 @@ mod integration_tests { downstairs.clone_region(source).await } + // Stop this downstairs, freeing the port it was listening on. Used + // to simulate a downstairs that is not running. The address that was + // assigned during spawn is still recorded in any CrucibleOpts we + // handed out, so the upstairs will try (and fail) to connect to it. + pub async fn stop(&mut self) -> Result<()> { + if let Some(downstairs) = self.downstairs.take() { + downstairs.stop().await?; + } + Ok(()) + } + pub fn address(&self) -> SocketAddr { // If start_downstairs returned Ok, then address will be populated self.downstairs.as_ref().unwrap().address() @@ -1160,6 +1171,61 @@ mod integration_tests { Ok(()) } + #[tokio::test] + async fn integration_test_just_read_one_downstairs() -> Result<()> { + // Create three read-only downstairs, but only leave one of them + // running. A single read-only downstairs is enough to activate a + // read-only upstairs, so both activation and a read should succeed. + const BLOCK_SIZE: usize = 512; + + // small(true) creates and starts all three downstairs read only. We + // capture the opts (which record all three addresses) before stopping + // two of them, leaving only downstairs1 running. + let mut tds = DefaultTestDownstairsSet::small(true).await?; + let opts = tds.opts(); + + tds.downstairs2.stop().await?; + tds.downstairs3.stop().await?; + + // Put the region under sub_volumes (not as a read_only_parent) so that + // flushes are actually sent to it. + let vcr = VolumeConstructionRequest::Volume { + id: Uuid::new_v4(), + block_size: BLOCK_SIZE as u64, + sub_volumes: vec![VolumeConstructionRequest::Region { + block_size: BLOCK_SIZE as u64, + blocks_per_extent: tds.blocks_per_extent(), + extent_count: tds.extent_count(), + opts, + generation: 1, + }], + read_only_parent: None, + }; + + let volume = Volume::construct(vcr, None, csl()).await?; + volume.activate().await?; + + // Read one block: should be all 0x00 + let mut buffer = Buffer::new(1, BLOCK_SIZE); + volume.read(BlockIndex(0), &mut buffer).await?; + + assert_eq!(vec![0x00; BLOCK_SIZE], &buffer[..]); + + // Manually send a flush. With only one downstairs running, the flush + // still completes: the two stopped downstairs have their jobs moved to + // Skipped, so the flush is complete on all clients and acks back to us. + // This should not hang. + volume.flush(None).await?; + + // A second read after the flush should also complete successfully. + let mut buffer = Buffer::new(1, BLOCK_SIZE); + volume.read(BlockIndex(0), &mut buffer).await?; + + assert_eq!(vec![0x00; BLOCK_SIZE], &buffer[..]); + + Ok(()) + } + #[tokio::test] async fn integration_test_volume_write_unwritten_1() -> Result<()> { // Test a simple single layer volume, verify write_unwritten diff --git a/upstairs/src/dummy_downstairs_tests.rs b/upstairs/src/dummy_downstairs_tests.rs index 7547f83de..78befc8fe 100644 --- a/upstairs/src/dummy_downstairs_tests.rs +++ b/upstairs/src/dummy_downstairs_tests.rs @@ -2654,20 +2654,12 @@ async fn test_no_read_only_live_repair() { assert!(matches!(harness.ds2.try_recv(), Err(TryRecvError::Empty))); assert!(matches!(harness.ds3.try_recv(), Err(TryRecvError::Empty))); - // Flush to clean out skipped jobs - { - // We must `spawn` here because `flush` will wait for the - // response to come back before returning - let jh = harness.spawn(|guest| async move { - guest.flush(None).await.unwrap(); - }); - - harness.ds2.ack_flush().await; - harness.ds3.ack_flush().await; - - // Wait for the flush to come back - jh.await.unwrap(); - } + // Flush to clean out skipped jobs. A read-only guest flush is acked + // locally and never sent to the downstairs, so we trigger the internal + // (automatic) flush instead, which does reach the downstairs. + harness.guest.flush_check().await.unwrap(); + harness.ds2.ack_flush().await; + harness.ds3.ack_flush().await; // Confirm that DS1 has been disconnected (and cannot reply to jobs) { diff --git a/upstairs/src/guest.rs b/upstairs/src/guest.rs index b8d24d684..9695eaccd 100644 --- a/upstairs/src/guest.rs +++ b/upstairs/src/guest.rs @@ -235,6 +235,16 @@ impl Guest { self.send_and_wait(|done| BlockOp::FaultDownstairs { client_id, done }) .await } + + /// Run the work that the automatic flush timer performs + /// + /// This is used in tests to deterministically trigger the internal flush + /// that a read-only guest flush intentionally skips. + #[cfg(test)] + pub async fn flush_check(&self) -> Result<(), CrucibleError> { + self.send_and_wait(|done| BlockOp::FlushCheck { done }) + .await + } } #[async_trait] diff --git a/upstairs/src/lib.rs b/upstairs/src/lib.rs index c2cea6a3f..38e129127 100644 --- a/upstairs/src/lib.rs +++ b/upstairs/src/lib.rs @@ -1576,6 +1576,11 @@ pub(crate) enum BlockOp { client_id: ClientId, done: BlockRes<()>, }, + + #[cfg(test)] + FlushCheck { + done: BlockRes<()>, + }, } /** diff --git a/upstairs/src/upstairs.rs b/upstairs/src/upstairs.rs index 9832cfe96..85e0092a6 100644 --- a/upstairs/src/upstairs.rs +++ b/upstairs/src/upstairs.rs @@ -1172,7 +1172,14 @@ impl Upstairs { done.send_err(CrucibleError::UpstairsInactive); return; } - + if self.cfg.read_only { + // While we ACK a guest sent flush here, The upstairs + // internally will still send a flush to all connected RO + // downstairs, which they are expected to handle. This + // internal flush serves to clean out completed jobs. + done.send_ok(()); + return; + } let n = self.downstairs.active_client_count(); let required = if snapshot_details.is_some() { 3 } else { 2 }; if n < required { @@ -1205,6 +1212,20 @@ impl Upstairs { ); done.send_ok(()); } + + #[cfg(test)] + BlockOp::FlushCheck { done } => { + // Deterministically run the work the automatic flush timer + // does, so a test can trigger the internal flush that a + // read-only guest flush intentionally skips. We omit the + // timer's has_jobs guard because the test is forcing this + // explicitly. + if self.need_flush { + let io_guard = self.try_acquire_io(0); + self.submit_flush(None, None, io_guard); + } + done.send_ok(()); + } } }