fix: --include-tables ignored in xmin sync, CTRL+C unresponsive during sync

- Fixed sync command ignoring --include-tables and other CLI filter flags
  when using xmin-based sync. Now skips interactive mode when CLI flags
  are provided and passes filtered tables to xmin sync.

- Fixed CTRL+C not responding during sync cycle. Daemon now checks
  shutdown signal during sync/reconciliation cycles using nested
  tokio::select! with biased polling.

Bump version to 7.0.4

Author: taariq

CHANGELOG.md

@@ -2,6 +2,14 @@
 
 All notable changes to this project will be documented in this file.
 
+## [7.0.4] - 2025-12-09
+
+### Fixed
+
+- **`--include-tables` ignored in xmin sync**: Fixed bug where `--include-tables` and other CLI filter flags were ignored by the `sync` command when using xmin-based sync. The sync command now correctly respects table filters when CLI flags are provided, skipping interactive mode and passing filtered tables to xmin sync.
+
+- **CTRL+C not responding during sync**: Fixed bug where CTRL+C would not interrupt a running sync cycle. The daemon now checks for shutdown signals during sync and reconciliation cycles, allowing graceful termination at any point.
+
 ## [7.0.3] - 2025-12-08
 
 ### Fixed

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "database-replicator"
-version = "7.0.3"
+version = "7.0.4"
 edition = "2021"
 license = "Apache-2.0"
 description = "Universal database-to-PostgreSQL replication CLI. Supports PostgreSQL, SQLite, MongoDB, and MySQL."

Cargo.toml

@@ -519,7 +519,13 @@ async fn main() -> anyhow::Result<()> {
             app_state.target_url = Some(resolved_target.clone());
             database_replicator::state::save(&app_state)?;
 
-            let filter = if !no_interactive {
+            // Check if CLI filter flags were provided (skip interactive if so)
+            let has_cli_filters = include_databases.is_some()
+                || exclude_databases.is_some()
+                || include_tables.is_some()
+                || exclude_tables.is_some();
+
+            let filter = if !no_interactive && !has_cli_filters {
                 // Interactive mode (default) - prompt user to select databases and tables
                 let (filter, rules) =
                     database_replicator::interactive::select_databases_and_tables(&source).await?;
@@ -636,12 +642,39 @@ async fn main() -> anyhow::Result<()> {
                 );
                 tracing::info!("Using xmin-based sync (no source configuration required)");
 
+                // Extract tables from filter for xmin sync
+                // Filter stores "db.table" format, we need just table names for the source db
+                let source_parts = database_replicator::utils::parse_postgres_url(&source)?;
+                let source_db = &source_parts.database;
+
+                let tables_to_sync: Option<Vec<String>> = filter.include_tables().map(|tables| {
+                    tables
+                        .iter()
+                        .filter_map(|qualified| {
+                            // Split "db.table" into parts
+                            let parts: Vec<&str> = qualified.splitn(2, '.').collect();
+                            if parts.len() == 2 {
+                                let (db, table) = (parts[0], parts[1]);
+                                // Only include tables from the source database
+                                if db == source_db {
+                                    Some(table.to_string())
+                                } else {
+                                    None
+                                }
+                            } else {
+                                // No dot, treat as plain table name
+                                Some(qualified.clone())
+                            }
+                        })
+                        .collect()
+                });
+
                 // Use CLI-provided intervals or defaults
                 xmin_sync(
                     source,
                     resolved_target,
                     "public".to_string(), // Default schema
-                    None,                 // Discover all tables
+                    tables_to_sync,       // Tables from filter
                     sync_interval,        // CLI: --sync-interval (default 60s)
                     reconcile_interval,   // CLI: --reconcile-interval (default 3600s)
                     1000,                 // Batch size

src/main.rs

@@ -244,25 +244,41 @@ impl SyncDaemon {
 
         loop {
             tokio::select! {
+                biased; // Check shutdown first
+
+                _ = shutdown.recv() => {
+                    tracing::info!("Shutdown signal received, stopping SyncDaemon");
+                    break;
+                }
                 _ = sync_interval.tick() => {
                     cycles += 1;
                     tracing::info!("Starting sync cycle {}", cycles);
 
-                    match self.run_sync_cycle().await {
-                        Ok(stats) => {
-                            tracing::info!(
-                                "Sync cycle {} completed: {} tables, {} rows in {}ms",
-                                cycles,
-                                stats.tables_synced,
-                                stats.rows_synced,
-                                stats.duration_ms
-                            );
-                            if !stats.errors.is_empty() {
-                                tracing::warn!("Sync cycle had {} errors", stats.errors.len());
-                            }
+                    // Run sync cycle with shutdown check - abort if shutdown received
+                    tokio::select! {
+                        biased;
+                        _ = shutdown.recv() => {
+                            tracing::info!("Shutdown signal received during sync cycle, aborting");
+                            break;
                         }
-                        Err(e) => {
-                            tracing::error!("Sync cycle {} failed: {}", cycles, e);
+                        result = self.run_sync_cycle() => {
+                            match result {
+                                Ok(stats) => {
+                                    tracing::info!(
+                                        "Sync cycle {} completed: {} tables, {} rows in {}ms",
+                                        cycles,
+                                        stats.tables_synced,
+                                        stats.rows_synced,
+                                        stats.duration_ms
+                                    );
+                                    if !stats.errors.is_empty() {
+                                        tracing::warn!("Sync cycle had {} errors", stats.errors.len());
+                                    }
+                                }
+                                Err(e) => {
+                                    tracing::error!("Sync cycle {} failed: {}", cycles, e);
+                                }
+                            }
                         }
                     }
                 }
@@ -276,25 +292,31 @@ impl SyncDaemon {
                     reconcile_cycles += 1;
                     tracing::info!("Starting reconciliation cycle {}", reconcile_cycles);
 
-                    match self.run_reconciliation().await {
-                        Ok(stats) => {
-                            tracing::info!(
-                                "Reconciliation cycle {} completed: {} tables, {} rows deleted in {}ms",
-                                reconcile_cycles,
-                                stats.tables_synced,
-                                stats.rows_deleted,
-                                stats.duration_ms
-                            );
+                    // Run reconciliation with shutdown check
+                    tokio::select! {
+                        biased;
+                        _ = shutdown.recv() => {
+                            tracing::info!("Shutdown signal received during reconciliation, aborting");
+                            break;
                         }
-                        Err(e) => {
-                            tracing::error!("Reconciliation cycle {} failed: {}", reconcile_cycles, e);
+                        result = self.run_reconciliation() => {
+                            match result {
+                                Ok(stats) => {
+                                    tracing::info!(
+                                        "Reconciliation cycle {} completed: {} tables, {} rows deleted in {}ms",
+                                        reconcile_cycles,
+                                        stats.tables_synced,
+                                        stats.rows_deleted,
+                                        stats.duration_ms
+                                    );
+                                }
+                                Err(e) => {
+                                    tracing::error!("Reconciliation cycle {} failed: {}", reconcile_cycles, e);
+                                }
+                            }
                         }
                     }
                 }
-                _ = shutdown.recv() => {
-                    tracing::info!("Shutdown signal received, stopping SyncDaemon");
-                    break;
-                }
             }
         }