chore: reformat

Author: PhotonQuantum

rsync-core/src/logging/either.rs

@@ -1,5 +1,5 @@
-use tracing_subscriber::fmt::writer::EitherWriter;
 use tracing_subscriber::fmt::MakeWriter;
+use tracing_subscriber::fmt::writer::EitherWriter;
 
 pub enum EitherMakeWriter<A, B> {
     Left(A),

rsync-core/src/logging/mod.rs

@@ -1,13 +1,13 @@
 use std::panic::PanicHookInfo;
 use std::{env, io, panic};
 
-use tracing::{error, Subscriber};
+use tracing::{Subscriber, error};
 use tracing_error::ErrorLayer;
 use tracing_subscriber::filter::filter_fn;
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::registry::LookupSpan;
 use tracing_subscriber::util::SubscriberInitExt;
-use tracing_subscriber::{reload, EnvFilter, Layer};
+use tracing_subscriber::{EnvFilter, Layer, reload};
 
 pub use opts::{LogFormat, LogTarget};
 

rsync-core/src/pg_lock.rs

@@ -1,5 +1,5 @@
-use blake2::digest::consts::U8;
 use blake2::Digest;
+use blake2::digest::consts::U8;
 use eyre::bail;
 use sqlx::PgConnection;
 

rsync-core/src/s3.rs

@@ -1,7 +1,7 @@
 use eyre::Result;
+use opendal::Operator;
 use opendal::layers::{RetryLayer, TimeoutLayer, TracingLayer};
 use opendal::services::S3;
-use opendal::Operator;
 
 #[derive(Debug, Clone)]
 pub struct S3Opts {

rsync-fetcher/src/main.rs

@@ -16,19 +16,19 @@ use sqlx::PgPool;
 use tokio::sync::mpsc;
 use tracing::info;
 
-use rsync_core::logging::{init_color_eyre, init_logger};
 use rsync_core::logging::{LogFormat, LogTarget};
+use rsync_core::logging::{init_color_eyre, init_logger};
 use rsync_core::pg::{
-    change_revision_status, create_revision, ensure_repository, insert_task, RevisionStatus,
+    RevisionStatus, change_revision_status, create_revision, ensure_repository, insert_task,
 };
 use rsync_core::pg_lock::PgLock;
-use rsync_core::s3::{build_operator, S3Opts};
+use rsync_core::s3::{S3Opts, build_operator};
 
 use crate::pg::{
     analyse_objects, create_fl_table, drop_fl_table, insert_file_list_to_db, update_parent_ids,
 };
 use crate::plan::diff_and_apply;
-use crate::rsync::{finalize, start_handshake, TaskBuilders};
+use crate::rsync::{TaskBuilders, finalize, start_handshake};
 use crate::utils::{flatten_err, plan_stat};
 
 #[global_allocator]

rsync-fetcher/src/pg.rs

@@ -155,7 +155,7 @@ mod tests {
     mod db_required {
         use sqlx::PgPool;
 
-        use rsync_core::pg::{create_revision, ensure_repository, FileType, RevisionStatus};
+        use rsync_core::pg::{FileType, RevisionStatus, create_revision, ensure_repository};
         use rsync_core::tests::generate_random_namespace;
 
         use crate::pg::update_parent_ids;

rsync-fetcher/src/plan.rs

@@ -172,13 +172,13 @@ mod tests {
 
         use rsync_core::metadata::Metadata;
         use rsync_core::pg::{
-            change_revision_status, create_revision, ensure_repository, FileType, RevisionStatus,
+            FileType, RevisionStatus, change_revision_status, create_revision, ensure_repository,
         };
         use rsync_core::tests::generate_random_namespace;
         use rsync_core::tests::insert_to_revision;
 
         use crate::pg::{create_fl_table, insert_file_list_to_db};
-        use crate::plan::{diff_and_apply, TransferItem};
+        use crate::plan::{TransferItem, diff_and_apply};
         use crate::rsync::file_list::FileEntry;
 
         async fn assert_entry_eq<'a>(

rsync-fetcher/src/rsync.rs

@@ -1,4 +1,4 @@
-use eyre::{bail, Context, ContextCompat, Result};
+use eyre::{Context, ContextCompat, Result, bail};
 use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
 use tokio::net::TcpStream;
 use tokio_socks::tcp::Socks5Stream;

rsync-fetcher/src/rsync/checksum.rs

@@ -204,7 +204,7 @@ mod tests {
     use tempfile::tempfile;
     use test_strategy::proptest;
 
-    use crate::rsync::checksum::{checksum_payload, checksum_payload_basic, SumHead};
+    use crate::rsync::checksum::{SumHead, checksum_payload, checksum_payload_basic};
 
     #[inline]
     fn must_checksum_payload_basic_eq_simd_(data: &[u8]) -> (Vec<u8>, Vec<u8>) {

rsync-fetcher/src/rsync/checksum/md4_simd/x86_simd_transpose.rs

@@ -3,9 +3,9 @@
 use arrayref::array_ref;
 
 use self::arch::{
-    __m128i, __m256i, _mm256_castsi128_si256, _mm256_inserti128_si256, _mm256_unpackhi_epi32,
-    _mm256_unpackhi_epi64, _mm256_unpacklo_epi32, _mm256_unpacklo_epi64, _mm_loadu_si128,
-    _mm_unpackhi_epi32, _mm_unpackhi_epi64, _mm_unpacklo_epi32, _mm_unpacklo_epi64,
+    __m128i, __m256i, _mm_loadu_si128, _mm_unpackhi_epi32, _mm_unpackhi_epi64, _mm_unpacklo_epi32,
+    _mm_unpacklo_epi64, _mm256_castsi128_si256, _mm256_inserti128_si256, _mm256_unpackhi_epi32,
+    _mm256_unpackhi_epi64, _mm256_unpacklo_epi32, _mm256_unpacklo_epi64,
 };
 #[cfg(target_arch = "x86")]
 use std::arch::x86 as arch;

rsync-fetcher/src/rsync/downloader.rs

@@ -6,9 +6,9 @@ use std::io::SeekFrom;
 use std::path::PathBuf;
 use std::sync::Arc;
 
-use eyre::{bail, eyre, Result};
+use eyre::{Result, bail, eyre};
 use futures::stream::FuturesUnordered;
-use futures::{FutureExt, TryStreamExt, SinkExt};
+use futures::{FutureExt, SinkExt, TryStreamExt};
 use opendal::Operator;
 use tokio::fs::{File, OpenOptions};
 use tokio::io::AsyncSeekExt;
@@ -126,7 +126,7 @@ impl Downloader {
             let permit = self.basis_tx.reserve().await?;
             self.pb.inc_basis(1);
             self.pb.inc_basis_downloading(1);
-            
+
             let basis_path = self
                 .basis_dir
                 .join(format!("{:x}", hash(entry.path).as_hex()));
@@ -138,15 +138,15 @@ impl Downloader {
                 .open(&basis_path)
                 .await?;
             let mut file_sink = FramedWrite::new(&mut basis_file, BytesCodec::new());
-            
+
             let key = format!("{}{:x}", self.s3_prefix, entry.blake2b_hash.as_hex());
             let copy_result = {
                 let file_sink_mut = &mut file_sink;
                 async move {
                     let reader = self.s3.reader_with(&key).chunk(DOWNLOAD_CHUNK_SIZE).await?;
                     let mut reader_stream = reader.into_bytes_stream(..).await?;
                     file_sink_mut.send_all(&mut reader_stream).await?;
-                    
+
                     Ok::<_, io::Error>(())
                 }
             }

rsync-fetcher/src/rsync/file_list.rs

@@ -6,7 +6,7 @@ use std::path::Path;
 use std::time::{Duration, SystemTime};
 
 use clean_path::Clean;
-use eyre::{bail, Result};
+use eyre::{Result, bail};
 use indicatif::ProgressBar;
 use tokio::io::{AsyncReadExt, BufReader};
 use tokio::net::tcp::OwnedReadHalf;

rsync-fetcher/src/rsync/generator.rs

@@ -6,15 +6,15 @@ use std::path::Path;
 use std::sync::Arc;
 
 use eyre::Result;
-use futures::{stream, StreamExt};
+use futures::{StreamExt, stream};
 use tokio::fs::File;
 use tokio::io::AsyncWriteExt;
 use tokio::net::tcp::OwnedWriteHalf;
-use tokio::sync::{mpsc, Semaphore};
+use tokio::sync::{Semaphore, mpsc};
 use tracing::{debug, info};
 
 use crate::plan::TransferItem;
-use crate::rsync::checksum::{checksum_payload, SumHead};
+use crate::rsync::checksum::{SumHead, checksum_payload};
 use crate::rsync::file_list::FileEntry;
 use crate::rsync::progress_display::ProgressDisplay;
 use crate::utils::ignore_mode;

rsync-fetcher/src/rsync/handshake.rs

@@ -5,22 +5,22 @@
 
 use std::fmt::{Debug, Formatter};
 
-use base64::engine::general_purpose;
 use base64::Engine;
+use base64::engine::general_purpose;
 use digest::Digest;
-use eyre::{bail, Result};
+use eyre::{Result, bail};
 use md4::Md4;
 use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
-use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 use tokio::net::TcpStream;
+use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 use tracing::{debug, instrument};
 use url::Url;
 use zeroize::{Zeroize, ZeroizeOnDrop};
 
 use crate::rsync::envelope::EnvelopeRead;
 use crate::rsync::filter::Rule;
 use crate::rsync::mux_conn::MuxConn;
-use crate::rsync::version::{Version, SUPPORTED_VERSION};
+use crate::rsync::version::{SUPPORTED_VERSION, Version};
 
 #[derive(Zeroize, ZeroizeOnDrop)]
 pub struct Auth {

rsync-fetcher/src/rsync/mux_conn.rs

@@ -6,19 +6,19 @@ use opendal::Operator;
 use tempfile::TempDir;
 use tokio::io::BufReader;
 use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
-use tokio::sync::{mpsc, Semaphore};
+use tokio::sync::{Semaphore, mpsc};
 
 use rsync_core::metadata::Metadata;
 
 use crate::consts::{BASIS_BUFFER_LIMIT, UPLOAD_CONN};
+use crate::rsync::TaskBuilders;
 use crate::rsync::downloader::Downloader;
 use crate::rsync::envelope::EnvelopeRead;
 use crate::rsync::file_list::FileEntry;
 use crate::rsync::generator::Generator;
 use crate::rsync::progress_display::ProgressDisplay;
 use crate::rsync::receiver::Receiver;
 use crate::rsync::uploader::Uploader;
-use crate::rsync::TaskBuilders;
 
 pub struct MuxConn {
     tx: OwnedWriteHalf,
@@ -27,7 +27,11 @@ pub struct MuxConn {
 }
 
 impl MuxConn {
-    pub const fn new(tx: OwnedWriteHalf, rx: EnvelopeRead<BufReader<OwnedReadHalf>>, seed: i32) -> Self {
+    pub const fn new(
+        tx: OwnedWriteHalf,
+        rx: EnvelopeRead<BufReader<OwnedReadHalf>>,
+        seed: i32,
+    ) -> Self {
         Self { tx, rx, seed }
     }
     pub async fn recv_file_list(&mut self) -> Result<Vec<FileEntry>> {

rsync-fetcher/src/rsync/progress_display.rs

@@ -1,7 +1,7 @@
 use std::fmt::{Display, Formatter};
 use std::ops::Deref;
-use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::Duration;
 
 use eyre::Result;

rsync-fetcher/src/rsync/receiver.rs

@@ -6,15 +6,15 @@ use std::ops::{Deref, DerefMut};
 use std::sync::Arc;
 
 use blake2::Blake2b;
-use digest::consts::U20;
 use digest::Digest;
-use eyre::{ensure, Result};
+use digest::consts::U20;
+use eyre::{Result, ensure};
 use md4::Md4;
-use tempfile::{tempfile_in, TempDir, TempPath};
+use tempfile::{TempDir, TempPath, tempfile_in};
 use tokio::fs;
 use tokio::io::{AsyncReadExt, BufReader};
 use tokio::net::tcp::OwnedReadHalf;
-use tokio::sync::{mpsc, Semaphore};
+use tokio::sync::{Semaphore, mpsc};
 use tracing::{debug, info, instrument};
 
 use rsync_core::utils::ToHex;

rsync-fetcher/src/rsync/stats.rs

@@ -1,4 +1,3 @@
-
 /// Stats returned by server at the end of transmission.
 #[derive(Debug, Copy, Clone)]
 #[allow(dead_code)]

rsync-fetcher/src/rsync/uploader.rs

@@ -16,7 +16,7 @@ use tokio::sync::mpsc;
 use tracing::{debug, info, warn};
 
 use rsync_core::metadata::{MetaExtra, Metadata};
-use rsync_core::utils::{ToHex, ATTR_CHAR};
+use rsync_core::utils::{ATTR_CHAR, ToHex};
 
 use crate::consts::UPLOAD_CONN;
 use crate::rsync::file_list::FileEntry;

rsync-fetcher/src/rsync/version.rs

@@ -1,6 +1,6 @@
 use std::fmt::Display;
 
-use eyre::{eyre, Context, Result};
+use eyre::{Context, Result, eyre};
 use scan_fmt::scan_fmt;
 use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use tracing::debug;

rsync-fetcher/src/utils.rs

@@ -3,8 +3,8 @@ use std::ops::Add;
 
 use blake2::Blake2b;
 use bytesize::ByteSize;
-use digest::consts::U20;
 use digest::Digest;
+use digest::consts::U20;
 use tracing::warn;
 
 use rsync_core::utils::ToHex;

rsync-gateway/src/app.rs

@@ -9,14 +9,14 @@ use futures::future;
 use opendal::Operator;
 use sqlx::PgPool;
 
-use rsync_core::s3::{build_operator, S3Opts};
+use rsync_core::s3::{S3Opts, build_operator};
 use rsync_core::utils::AbortJoinHandle;
 
 use crate::cache::NSCache;
 use crate::handler;
 use crate::listener::RevisionsChangeListener;
 use crate::opts::{Config, Endpoint};
-use crate::state::{listen_for_updates, State};
+use crate::state::{State, listen_for_updates};
 
 pub struct Prefix(pub String);
 
@@ -45,7 +45,10 @@ pub async fn configure<B: for<'a> Fn(&'a Config, &'a Endpoint) -> Result<Operato
     opts: &Config,
     op_builder: B,
     pool: PgPool,
-) -> Result<(AbortJoinHandle<()>, impl Fn(&mut ServiceConfig) + Clone + use<B>)> {
+) -> Result<(
+    AbortJoinHandle<()>,
+    impl Fn(&mut ServiceConfig) + Clone + use<B>,
+)> {
     let listener = RevisionsChangeListener::default();
     let listener_handle = listener.spawn(&pool);
 
@@ -74,29 +77,26 @@ pub async fn configure<B: for<'a> Fn(&'a Config, &'a Endpoint) -> Result<Operato
         .await?,
     );
 
-    Ok((
-        listener_handle,
-        move |cfg: &mut ServiceConfig| {
-            for (prefix, endpoint, revision, guard, cache, op) in &*prefix_state {
-                let state = State::new(revision.clone(), guard.clone());
-                cfg.service(
-                    web::scope(&format!("/{prefix}"))
-                        .app_data(Data::new(Prefix(prefix.clone())))
-                        .app_data(Data::new(state))
-                        .app_data(Data::from(endpoint.clone()))
-                        .app_data(Data::from(cache.clone()))
-                        .app_data(Data::new(op.clone()))
-                        .route(
-                            "/_revisions",
-                            web::route().guard(get_or_head()).to(handler::rev_handler),
-                        )
-                        .service(
-                            web::resource(["", "/{path:(.|/)*}"])
-                                .route(web::route().guard(get_or_head()).to(handler::main_handler)),
-                        ),
-                );
-            }
-            cfg.app_data(Data::new(pool.clone()));
-        },
-    ))
-}
+    Ok((listener_handle, move |cfg: &mut ServiceConfig| {
+        for (prefix, endpoint, revision, guard, cache, op) in &*prefix_state {
+            let state = State::new(revision.clone(), guard.clone());
+            cfg.service(
+                web::scope(&format!("/{prefix}"))
+                    .app_data(Data::new(Prefix(prefix.clone())))
+                    .app_data(Data::new(state))
+                    .app_data(Data::from(endpoint.clone()))
+                    .app_data(Data::from(cache.clone()))
+                    .app_data(Data::new(op.clone()))
+                    .route(
+                        "/_revisions",
+                        web::route().guard(get_or_head()).to(handler::rev_handler),
+                    )
+                    .service(
+                        web::resource(["", "/{path:(.|/)*}"])
+                            .route(web::route().guard(get_or_head()).to(handler::main_handler)),
+                    ),
+            );
+        }
+        cfg.app_data(Data::new(pool.clone()));
+    }))
+}