style: apply cargo fmt formatting

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: YuweiXiao

duckpipe-core/src/duckdb_flush.rs

@@ -113,14 +113,12 @@ fn discover_lake_table_info(
 
     let mut column_types = Vec::with_capacity(expected_attnames.len());
     for name in expected_attnames {
-        let dtype = lake_col_map
-            .get(&name.to_lowercase())
-            .ok_or_else(|| {
-                format!(
-                    "column '{}' not found in DuckLake table {}.{}",
-                    name, lake_schema, target_table
-                )
-            })?;
+        let dtype = lake_col_map.get(&name.to_lowercase()).ok_or_else(|| {
+            format!(
+                "column '{}' not found in DuckLake table {}.{}",
+                name, lake_schema, target_table
+            )
+        })?;
         column_types.push(dtype.clone());
     }
 
@@ -175,7 +173,7 @@ impl FlushWorker {
         db.execute_batch(
             "SET ducklake_retry_wait_ms = 100; \
              SET ducklake_retry_backoff = 2.0; \
-             SET ducklake_max_retry_count = 10;"
+             SET ducklake_max_retry_count = 10;",
         )
         .map_err(|e| format!("duckdb set retry: {}", e))?;
 
@@ -236,10 +234,7 @@ impl FlushWorker {
         // pgoutput always includes every column value in UPDATE WAL records.
         // Any 'u'-status (TOAST unchanged) column reaching the flush path means
         // the source table had its REPLICA IDENTITY changed after add_table().
-        if changes
-            .iter()
-            .any(|c| c.col_unchanged.iter().any(|&u| u))
-        {
+        if changes.iter().any(|c| c.col_unchanged.iter().any(|&u| u)) {
             return Err(
                 "TOAST unchanged column detected in WAL — source table must have \
                  REPLICA IDENTITY FULL. Run: ALTER TABLE <name> REPLICA IDENTITY FULL"
@@ -403,12 +398,10 @@ impl FlushWorker {
         let deleted_count: usize = if skip_delete {
             0
         } else {
-            self.db
-                .execute(&delete_sql, [])
-                .map_err(|e| {
-                    let _ = self.db.execute_batch("ROLLBACK");
-                    format!("duckdb delete from {}: {}", target_key, e)
-                })?
+            self.db.execute(&delete_sql, []).map_err(|e| {
+                let _ = self.db.execute_batch("ROLLBACK");
+                format!("duckdb delete from {}: {}", target_key, e)
+            })?
         };
         let t_delete_ms = t_phase.elapsed().as_secs_f64() * 1000.0;
 
@@ -427,12 +420,10 @@ impl FlushWorker {
             target_ref = target_ref,
             cols = all_cols.join(", ")
         );
-        self.db
-            .execute_batch(&insert_sql)
-            .map_err(|e| {
-                let _ = self.db.execute_batch("ROLLBACK");
-                format!("duckdb insert into {}: {}", target_key, e)
-            })?;
+        self.db.execute_batch(&insert_sql).map_err(|e| {
+            let _ = self.db.execute_batch("ROLLBACK");
+            format!("duckdb insert into {}: {}", target_key, e)
+        })?;
         let t_insert_ms = t_phase.elapsed().as_secs_f64() * 1000.0;
 
         let t_phase = Instant::now();
@@ -486,7 +477,6 @@ impl FlushWorker {
     }
 }
 
-
 /// Result of a DuckDB-based flush.
 #[derive(Debug)]
 pub struct DuckDbFlushResult {

duckpipe-core/src/flush_coordinator.rs

@@ -160,18 +160,19 @@ impl FlushCoordinator {
     ) {
         // Seed in-memory LSN from the persistent PG value if we haven't seen this table yet.
         // `or_insert` preserves any higher value already tracked from a completed flush.
-        self.per_table_lsn.entry(mapping_id).or_insert(initial_applied_lsn);
+        self.per_table_lsn
+            .entry(mapping_id)
+            .or_insert(initial_applied_lsn);
         // Cache target_key → mapping_id for lock-free lookup in get_min_applied_lsn_in_coordinator.
-        self.target_to_mapping.insert(target_key.to_string(), mapping_id);
+        self.target_to_mapping
+            .insert(target_key.to_string(), mapping_id);
         // Check if entry exists and thread is alive
         let needs_spawn = match self.threads.get(target_key) {
             None => true,
-            Some(entry) => {
-                match &entry.join_handle {
-                    Some(h) => h.is_finished(),
-                    None => true,
-                }
-            }
+            Some(entry) => match &entry.join_handle {
+                Some(h) => h.is_finished(),
+                None => true,
+            },
         };
 
         if !needs_spawn {
@@ -264,7 +265,9 @@ impl FlushCoordinator {
                 guard.last_lsn = change.lsn;
             }
             guard.changes.push(change);
-            self.backpressure.total_queued.fetch_add(1, Ordering::Relaxed);
+            self.backpressure
+                .total_queued
+                .fetch_add(1, Ordering::Relaxed);
             entry.queue_handle.condvar.notify_one();
         }
     }
@@ -413,7 +416,12 @@ impl FlushCoordinator {
     pub fn collect_results(&mut self) -> Vec<FlushThreadResult> {
         let mut results = Vec::new();
         while let Ok(r) = self.result_rx.try_recv() {
-            if let FlushThreadResult::Success { mapping_id, last_lsn, .. } = &r {
+            if let FlushThreadResult::Success {
+                mapping_id,
+                last_lsn,
+                ..
+            } = &r
+            {
                 let entry = self.per_table_lsn.entry(*mapping_id).or_insert(0);
                 if *last_lsn > *entry {
                     *entry = *last_lsn;
@@ -539,7 +547,9 @@ fn flush_thread_main(
             if control.shutdown.load(Ordering::Acquire) {
                 // Drop in-memory changes — confirmed_lsn was never advanced past them,
                 // so PostgreSQL will re-deliver them on the next startup.
-                backpressure.total_queued.fetch_sub(accumulated_count, Ordering::Relaxed);
+                backpressure
+                    .total_queued
+                    .fetch_sub(accumulated_count, Ordering::Relaxed);
                 pending_local.store(0, Ordering::Relaxed);
                 if control.drain_requested.load(Ordering::Acquire) {
                     signal_drain_complete(&drain_complete, &control);
@@ -622,7 +632,9 @@ fn flush_thread_main(
                 );
             }
             accumulated.clear();
-            backpressure.total_queued.fetch_sub(accumulated_count, Ordering::Relaxed);
+            backpressure
+                .total_queued
+                .fetch_sub(accumulated_count, Ordering::Relaxed);
             accumulated_count = 0;
             pending_local.store(0, Ordering::Relaxed);
             accumulated_lsn = 0;
@@ -697,7 +709,8 @@ fn do_flush(
             )) {
                 tracing::error!(
                     "pg_duckpipe: metrics update failed for {}: {}",
-                    result.target_key, e
+                    result.target_key,
+                    e
                 );
             }
             // Clear error state on success

duckpipe-core/src/flush_worker.rs

@@ -72,7 +72,9 @@ pub async fn update_error_state(
                 .await;
             tracing::warn!(
                 "pg_duckpipe: mapping {} transitioned to ERRORED after {} failures, retry in {}s",
-                mapping_id, count, backoff
+                mapping_id,
+                count,
+                backoff
             );
         }
     }
@@ -84,10 +86,7 @@ pub async fn update_error_state(
 }
 
 /// Clear error state on successful flush: reset consecutive_failures and error_message.
-pub async fn clear_error_on_success(
-    connstr: &str,
-    mapping_id: i32,
-) -> Result<(), String> {
+pub async fn clear_error_on_success(connstr: &str, mapping_id: i32) -> Result<(), String> {
     let (client, connection) = tokio_postgres::connect(connstr, NoTls)
         .await
         .map_err(|e| format!("clear_error connect: {}", e))?;

duckpipe-core/src/lib.rs

@@ -1,9 +1,9 @@
 pub mod decoder;
-pub mod log;
 pub mod duckdb_flush;
 pub mod error;
 pub mod flush_coordinator;
 pub mod flush_worker;
+pub mod log;
 pub mod metadata;
 pub mod queue;
 pub mod service;

duckpipe-core/src/metadata.rs

@@ -4,7 +4,7 @@ use std::collections::HashMap;
 
 use tokio_postgres::Client;
 
-use crate::types::{SyncGroup, TableMapping, format_lsn, parse_lsn};
+use crate::types::{format_lsn, parse_lsn, SyncGroup, TableMapping};
 
 /// Async metadata client wrapping a tokio-postgres connection.
 pub struct MetadataClient<'a> {
@@ -35,9 +35,7 @@ impl<'a> MetadataClient<'a> {
                 let publication: String = row.get(2);
                 let slot_name: String = row.get(3);
                 let confirmed_lsn_str: Option<String> = row.get(4);
-                let confirmed_lsn = confirmed_lsn_str
-                    .map(|s| parse_lsn(&s))
-                    .unwrap_or(0);
+                let confirmed_lsn = confirmed_lsn_str.map(|s| parse_lsn(&s)).unwrap_or(0);
                 SyncGroup {
                     id,
                     name,
@@ -81,16 +79,12 @@ impl<'a> MetadataClient<'a> {
         let target_table: String = row.get(4);
         let state: String = row.get(5);
         let snapshot_lsn_str: Option<String> = row.get(6);
-        let snapshot_lsn = snapshot_lsn_str
-            .map(|s| parse_lsn(&s))
-            .unwrap_or(0);
+        let snapshot_lsn = snapshot_lsn_str.map(|s| parse_lsn(&s)).unwrap_or(0);
         let enabled: bool = row.get(7);
         let source_oid: Option<i64> = row.get(8);
         let error_message: Option<String> = row.get(9);
         let applied_lsn_str: Option<String> = row.get(10);
-        let applied_lsn = applied_lsn_str
-            .map(|s| parse_lsn(&s))
-            .unwrap_or(0);
+        let applied_lsn = applied_lsn_str.map(|s| parse_lsn(&s)).unwrap_or(0);
 
         Ok(Some(TableMapping {
             id,
@@ -247,10 +241,7 @@ impl<'a> MetadataClient<'a> {
     }
 
     /// Auto-retry: transition ERRORED table back to STREAMING, clear error and failures.
-    pub async fn retry_errored_table(
-        &self,
-        mapping_id: i32,
-    ) -> Result<(), tokio_postgres::Error> {
+    pub async fn retry_errored_table(&self, mapping_id: i32) -> Result<(), tokio_postgres::Error> {
         self.client
             .execute(
                 "UPDATE duckpipe.table_mappings SET state = 'STREAMING', \
@@ -376,10 +367,7 @@ impl<'a> MetadataClient<'a> {
     /// Get the minimum applied_lsn across all active (STREAMING/CATCHUP) tables in a group.
     /// Returns 0 if any active table has NULL applied_lsn (not yet flushed), or if
     /// there are no active tables.
-    pub async fn get_min_applied_lsn(
-        &self,
-        group_id: i32,
-    ) -> Result<u64, tokio_postgres::Error> {
+    pub async fn get_min_applied_lsn(&self, group_id: i32) -> Result<u64, tokio_postgres::Error> {
         // For CATCHUP tables with NULL applied_lsn, use snapshot_lsn as the
         // effective floor: everything up to snapshot_lsn is either in the
         // snapshot copy or will be skipped by CATCHUP skip logic, so the slot

duckpipe-core/src/queue.rs

@@ -58,4 +58,3 @@ impl TableQueue {
         self.changes.is_empty()
     }
 }
-