Fast ack a flush to a RO volume.

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

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)
     {

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]

upstairs/src/lib.rs

@@ -1576,6 +1576,11 @@ pub(crate) enum BlockOp {
         client_id: ClientId,
         done: BlockRes<()>,
     },
+
+    #[cfg(test)]
+    FlushCheck {
+        done: BlockRes<()>,
+    },
 }
 
 /**

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(());
+            }
         }
     }