Min-quorum negotiation (again)

integration_tests/src/lib.rs

@@ -501,7 +501,24 @@ mod integration_tests {
             Ok(())
         }
 
+        /// Stops the downstairs task, return a `(port, rport)` tuple
+        pub async fn stop(&mut self) -> Result<(u16, u16)> {
+            let ds = self.downstairs.take().unwrap();
+            let port = ds.address().port();
+            let rport = ds.repair_address().port();
+            ds.stop().await?;
+            Ok((port, rport))
+        }
+
         pub async fn reboot_read_write(&mut self) -> Result<()> {
+            self.reboot_read_write_with_ports(0, 0).await
+        }
+
+        pub async fn reboot_read_write_with_ports(
+            &mut self,
+            port: u16,
+            rport: u16,
+        ) -> Result<()> {
             let downstairs = with_max_message_version(
                 Downstairs::new_builder(&self.path()?, false),
                 self.max_message_version,
@@ -514,6 +531,8 @@ mod integration_tests {
                     downstairs,
                     DownstairsClientSettings {
                         address: self.address,
+                        port,
+                        rport,
                         ..DownstairsClientSettings::default()
                     },
                 )
@@ -540,17 +559,6 @@ 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()
@@ -750,6 +758,19 @@ mod integration_tests {
             self.extent_count
         }
 
+        /// Activates an upstairs with all three downstairs, then deactivates
+        ///
+        /// Each downstairs records that it has been active, so it can count
+        /// toward min-quorum.  This uses generation 1, so later activations
+        /// must use generation 2 or higher.
+        pub async fn activate_all_once(&self) -> Result<()> {
+            let (guest, io) = Guest::new(None);
+            let _join_handle = up_main(self.opts(), 1, None, io, None)?;
+            guest.activate().await?;
+            guest.deactivate().await?;
+            Ok(())
+        }
+
         pub async fn reboot_read_only(&mut self) -> Result<()> {
             self.downstairs1.reboot_read_only().await?;
             self.downstairs2.reboot_read_only().await?;
@@ -3863,7 +3884,7 @@ mod integration_tests {
     #[tokio::test]
     async fn integration_test_volume_inactive_replace_downstairs() -> Result<()>
     {
-        // Replace a downstairs before the volume is active.
+        // Replace a downstairs before the volume is active. This should fail
         const BLOCK_SIZE: usize = 512;
 
         // Create three downstairs.
@@ -3895,10 +3916,13 @@ mod integration_tests {
                 test_downstairs_set.downstairs1_address(),
                 new_downstairs.address(),
             )
-            .await
-            .unwrap();
+            .await;
 
-        assert_eq!(res, ReplaceResult::Started);
+        assert!(res.is_err());
+        assert!(matches!(
+            res.err().unwrap(),
+            CrucibleError::ReplaceRequestInvalid(_)
+        ));
         Ok(())
     }
 
@@ -4490,9 +4514,8 @@ mod integration_tests {
 
     #[tokio::test]
     async fn integration_test_guest_replace_ds_before_active() -> Result<()> {
-        // Test using the guest layer to verify we can replace a downstairs
-        // before the upstairs is active.
-        const BLOCK_SIZE: usize = 512;
+        // Test using the guest layer to verify replacing a downstairs
+        // before the upstairs is active will fail.
 
         // Spin off three downstairs, build our Crucible struct.
         let tds = DefaultTestDownstairsSet::small(false).await?;
@@ -4512,26 +4535,13 @@ mod integration_tests {
                 tds.downstairs1_address(),
                 new_downstairs.address(),
             )
-            .await
-            .unwrap();
-
-        assert_eq!(res, ReplaceResult::Started);
-
-        guest.activate().await?;
-
-        // Write data in
-        guest
-            .write(
-                BlockIndex(0),
-                BytesMut::from(vec![0x55; BLOCK_SIZE * 10].as_slice()),
-            )
-            .await?;
-
-        // Read back our block post replacement, verify contents
-        let mut buffer = Buffer::new(10, BLOCK_SIZE);
-        guest.read(BlockIndex(0), &mut buffer).await?;
+            .await;
 
-        assert_eq!(vec![0x55_u8; BLOCK_SIZE * 10], &buffer[..]);
+        assert!(res.is_err());
+        assert!(matches!(
+            res.err().unwrap(),
+            CrucibleError::ReplaceRequestInvalid(_)
+        ));
 
         Ok(())
     }
@@ -6522,6 +6532,174 @@ mod integration_tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn connect_two_ds_then_deactivate() {
+        const BLOCK_SIZE: usize = 512;
+
+        // Spin off three downstairs, build our Crucible struct.
+        let mut tds = DefaultTestDownstairsSet::small(false).await.unwrap();
+
+        // Activate with all three once, so each has been active and can
+        // count toward min-quorum.
+        tds.activate_all_once().await.unwrap();
+        let opts = tds.opts();
+        tds.downstairs1.stop().await.unwrap();
+        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+
+        let (guest, io) = Guest::new(None);
+        let _join_handle = up_main(opts, 2, None, io, None).unwrap();
+        guest.activate().await.unwrap();
+
+        let res = guest
+            .write(
+                BlockIndex(0),
+                BytesMut::from(vec![0x55; BLOCK_SIZE * 2].as_slice()),
+            )
+            .await;
+        assert!(res.is_ok());
+
+        guest.deactivate().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn connect_two_ds_then_another() {
+        const BLOCK_SIZE: usize = 512;
+
+        // Spin off three downstairs, build our Crucible struct.
+        let mut tds = DefaultTestDownstairsSet::small(false).await.unwrap();
+
+        // Activate with all three once, so each has been active and can
+        // count toward min-quorum.
+        tds.activate_all_once().await.unwrap();
+        let opts = tds.opts();
+        let (ds1_port, ds1_rport) = tds.downstairs1.stop().await.unwrap();
+        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+
+        let (guest, io) = Guest::new(None);
+        let _join_handle = up_main(opts, 2, None, io, None).unwrap();
+        guest.activate().await.unwrap();
+
+        let res = guest
+            .write(
+                BlockIndex(0),
+                BytesMut::from(vec![0x55; BLOCK_SIZE * 2].as_slice()),
+            )
+            .await;
+        assert!(res.is_ok());
+
+        // Restart downstairs1, which should use live-repair to join the quorum
+        //
+        // We have to wait a while here, because there's a 10-second reconnect
+        // delay.
+        tds.downstairs1
+            .reboot_read_write_with_ports(ds1_port, ds1_rport)
+            .await
+            .unwrap();
+        tokio::time::sleep(tokio::time::Duration::from_secs(15)).await;
+        guest.deactivate().await.unwrap();
+
+        // Reconnect with only ds1 running, then confirm that it received the
+        // writes.  We'll come up in read-only mode so that we can connect with
+        // just a single Downstairs, to make sure the reads go to DS1.
+        tds.downstairs1.reboot_read_only().await.unwrap();
+        tds.downstairs2.stop().await.unwrap();
+        tds.downstairs3.stop().await.unwrap();
+        tds.crucible_opts.read_only = true;
+        tds.crucible_opts.target[0] = tds.downstairs1.address();
+        let opts = tds.opts();
+        let (guest, io) = Guest::new(None);
+        let _join_handle = up_main(opts, 1, None, io, None).unwrap();
+        guest.activate().await.unwrap();
+        let mut buf = Buffer::new(2, BLOCK_SIZE);
+        guest.read(BlockIndex(0), &mut buf).await.unwrap();
+
+        assert_eq!(buf.to_vec(), vec![0x55; BLOCK_SIZE * 2]);
+    }
+
+    #[tokio::test]
+    async fn min_quorum_live_repair() {
+        const BLOCK_SIZE: usize = 512;
+
+        // Spin off three downstairs, build our Crucible struct.
+        let mut tds = DefaultTestDownstairsSet::small(false).await.unwrap();
+
+        // Activate with all three once, so each has been active and can
+        // count toward min-quorum.
+        tds.activate_all_once().await.unwrap();
+
+        // Stop downstairs 1 before constructing the guest, so it won't be
+        // included and we'll do min-quorum reconciliation.
+        let (port, rport) = tds.downstairs1.stop().await.unwrap();
+
+        // Start the guest and do a write to ds 2 and 3.
+        let (guest, io) = Guest::new(None);
+        let opts = tds.opts();
+        let _join_handle = up_main(opts, 2, None, io, None).unwrap();
+        guest.activate().await.unwrap();
+        let res = guest
+            .write(
+                BlockIndex(0),
+                BytesMut::from(vec![0x55; BLOCK_SIZE * 2].as_slice()),
+            )
+            .await;
+        assert!(res.is_ok());
+
+        // Deactivate the guest, all without downstairs 1 participating
+        guest.deactivate().await.unwrap();
+
+        // At this point, the data has been written to DS 2 and 3.  We'll start
+        // up again with DS 1 and 2, so min-quorum should do reconciliation.
+
+        tds.downstairs1
+            .reboot_read_write_with_ports(port, rport)
+            .await
+            .unwrap();
+        tds.downstairs2.stop().await.unwrap();
+        guest.activate_with_gen(3).await.unwrap();
+
+        let mut buf = Buffer::new(2, BLOCK_SIZE);
+        guest.read(BlockIndex(0), &mut buf).await.unwrap();
+
+        assert_eq!(buf.to_vec(), vec![0x55; BLOCK_SIZE * 2]);
+    }
+
+    #[tokio::test]
+    async fn min_quorum_cancel() {
+        // Spin off three downstairs, build our Crucible struct.
+        let mut tds = DefaultTestDownstairsSet::small(false).await.unwrap();
+
+        // Activate with all three once, so each has been active and can
+        // count toward min-quorum.
+        tds.activate_all_once().await.unwrap();
+
+        // Stop downstairs 1 before constructing the guest, so it won't be
+        // included and we'll do min-quorum reconciliation.
+        let (port, rport) = tds.downstairs1.stop().await.unwrap();
+
+        // Start the guest and do a write to ds 2 and 3.
+        let (guest, io) = Guest::new(None);
+        let opts = tds.opts();
+        let _join_handle = up_main(opts, 2, None, io, None).unwrap();
+        let s = tokio::spawn(async move { guest.activate().await });
+
+        // Get into our min-quorum wait, which is 500 ms
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+
+        // Stop DS2
+        tds.downstairs2.stop().await.unwrap();
+
+        // Wait for the min-quorum timer to go off; it shouldn't panic!
+        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+
+        // Restart DS1, we're now eligible for min-quorum negotiation again
+        tds.downstairs1
+            .reboot_read_write_with_ports(port, rport)
+            .await
+            .unwrap();
+
+        s.await.unwrap().unwrap()
+    }
+
     /// Validate the VolumeInfo from the Pantry
     #[tokio::test]
     async fn test_pantry_get_volume_info() {

tools/make-nightly.sh

@@ -29,7 +29,6 @@ tar cavf out/crucible-nightly.tar.gz \
   tools/test_mem.sh \
   tools/test_replay.sh \
   tools/test_repair.sh \
-  tools/test_replace_special.sh \
   tools/test_restart_repair.sh \
   tools/test_nightly.sh \
   tools/loop-repair.sh \

tools/test_nightly.sh

@@ -98,20 +98,6 @@ else
     exit 1
 fi
 echo ""
-
-sleep 1
-banner replace
-banner special
-echo "$(date) test_replace_special start" >> "$output_file"
-./tools/test_replace_special.sh -l 50
-res=$?
-if [[ "$res" -eq 0 ]]; then
-    echo "$(date) test_replace_special pass" >> "$output_file"
-else
-    echo "$(date) test_replace_special fail with: $res" >> "$output_file"
-    (( err += 1 ))
-    exit 1
-fi
 duration=$SECONDS
 
 banner results

tools/test_repair.sh

@@ -213,6 +213,9 @@ while [[ $count -lt $loops ]]; do
         ds2_pid=$!
     fi
 
+    # Wait for it to start up
+    sleep 10
+
     cp "$verify_file" ${verify_file}.last
     echo "Verifying data now"
     echo ${ct} verify ${target_args} --verify-out "$verify_file" --verify-in "$verify_file" --range -q -g "$generation" > "$test_log"