deduplicate read_change_counter

Author: neubaner

bottomless/src/replicator.rs

@@ -607,7 +607,9 @@ impl Replicator {
             return Ok(());
         }
         tracing::debug!("Snapshotting {}", self.db_path);
-        let change_counter = match self.use_compression {
+
+        let mut reader = tokio::fs::File::open(&self.db_path).await?;
+        match self.use_compression {
             CompressionKind::None => {
                 self.client
                     .put_object()
@@ -616,11 +618,8 @@ impl Replicator {
                     .body(ByteStream::from_path(&self.db_path).await?)
                     .send()
                     .await?;
-                let mut reader = tokio::fs::File::open(&self.db_path).await?;
-                Self::read_change_counter(&mut reader).await?
             }
             CompressionKind::Gzip => {
-                let mut reader = tokio::fs::File::open(&self.db_path).await?;
                 let buf_reader = tokio::io::BufReader::new(reader.try_clone().await?);
                 let gzip_reader = async_compression::tokio::bufread::GzipEncoder::new(buf_reader);
 
@@ -630,8 +629,6 @@ impl Replicator {
                 // the whole snapshot in memory because S3 requires the `Content-Length` header
                 // to be set.
                 upload_s3_multipart(&self.client, &key, &self.bucket, gzip_reader).await?;
-
-                Self::read_change_counter(&mut reader).await?
             }
         };
 
@@ -641,6 +638,7 @@ impl Replicator {
          ** incremented on each transaction in WAL mode."
          ** Instead, we need to consult WAL checksums.
          */
+        let change_counter = Self::read_change_counter(&mut reader).await?;
         let change_counter_key = format!("{}-{}/.changecounter", self.db_name, self.generation);
         self.client
             .put_object()