Send deactivate flush if need_flush is set (#1573)

Otherwise, deactivate can fail in a case where there are no active jobs
but `need_flush` is set: it will deactivate right away (because there
are no active jobs), then the automatic flush will be sent to a
Downstairs in an invalid state (`Deactivated` or `New`).

This PR will hopefully fix
#1571

integration_tests/Cargo.toml

@@ -19,7 +19,7 @@ crucible-client-types.workspace = true
 crucible-downstairs = { workspace = true, features = ["integration-tests"] }
 crucible-pantry-client.workspace = true
 crucible-pantry.workspace = true
-crucible.workspace = true
+crucible = { workspace = true, features = ["integration-tests"] }
 dropshot.workspace = true
 futures-core.workspace = true
 futures.workspace = true

integration_tests/src/lib.rs

@@ -5809,4 +5809,41 @@ mod test {
             }
         ));
     }
+
+    #[tokio::test]
+    async fn test_auto_flush_deactivate() {
+        let log = csl();
+        let child = TestDownstairsSet::small(false).await.unwrap();
+        let vcr = VolumeConstructionRequest::Volume {
+            id: Uuid::new_v4(),
+            block_size: 512,
+            sub_volumes: vec![VolumeConstructionRequest::Region {
+                block_size: 512,
+                blocks_per_extent: child.blocks_per_extent(),
+                extent_count: child.extent_count(),
+                opts: child.opts(),
+                gen: 2,
+            }],
+            read_only_parent: None,
+        };
+
+        let volume = Volume::construct(vcr, None, log.clone()).await.unwrap();
+        volume.activate().await.unwrap();
+
+        // Send exactly IO_CACHED_MAX_JOBS to force a Barrier to be sent.  The
+        // barrier empties out the active list, so deactivation may proceed.
+        for _ in 0..crucible::testing::IO_CACHED_MAX_JOBS {
+            let mut buf = Buffer::new(2, 512);
+            volume.read(BlockIndex(0), &mut buf).await.unwrap();
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+
+        // At this point, need_flush is set, but there are no active jobs.
+        // Deactivation must be aware of the need_flush flag; otherwise, the
+        // Upstairs will attempt to submit that flush after deactivation.
+        volume.deactivate().await.unwrap();
+
+        // Make sure everything worked
+        volume.activate().await.unwrap();
+    }
 }

upstairs/Cargo.toml

@@ -12,6 +12,7 @@ path = "src/lib.rs"
 [features]
 asm = ["usdt/asm"]
 notify-nexus = ["nexus-client", "internal-dns", "progenitor-client", "http", "omicron-uuid-kinds"]
+integration-tests = []
 
 [dependencies]
 anyhow.workspace = true

upstairs/src/lib.rs

@@ -110,6 +110,12 @@ const IO_CACHED_MAX_BYTES: u64 = 1024 * 1024 * 1024; // 1 GiB
 /// them comes back.
 const IO_CACHED_MAX_JOBS: u64 = 10000;
 
+// Re-exports for unit testing
+#[cfg(feature = "integration-tests")]
+pub mod testing {
+    pub const IO_CACHED_MAX_JOBS: u64 = super::IO_CACHED_MAX_JOBS;
+}
+
 /// The BlockIO trait behaves like a physical NVMe disk (or a virtio virtual
 /// disk): there is no contract about what order operations that are submitted
 /// between flushes are performed in.

upstairs/src/upstairs.rs

@@ -1278,7 +1278,7 @@ impl Upstairs {
             }
             UpstairsState::Active => (),
         }
-        if !self.downstairs.can_deactivate_immediately() {
+        if self.need_flush || !self.downstairs.can_deactivate_immediately() {
             debug!(self.log, "not ready to deactivate; submitting final flush");
             let io_guard = self.try_acquire_io(0);
             self.submit_flush(None, None, io_guard);