update-pg-walstream-0.4.0

.cargo/audit.toml

@@ -1,6 +1,5 @@
 [advisories]
 ignore = [
-    "RUSTSEC-2023-0071",
-    "RUSTSEC-2026-0001"
+    "RUSTSEC-2023-0071"
 ]
 

Cargo.toml

@@ -28,8 +28,8 @@ thiserror = "2.0.12"
 lazy_static = "1.5"
 tracing = "0.1.41"
 tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }
-pg_walstream = "0.3.0"
-tiberius = { version = "0.12", features = ["tds73", "sql-browser-tokio", "bigdecimal", "rust_decimal", "time", "chrono"] }
+pg_walstream = "0.4.0"
+tiberius = { version = "0.12.3", features = ["tds73", "sql-browser-tokio", "bigdecimal", "rust_decimal", "time", "chrono"] }
 sqlx = { version = "0.8.6", features = ["runtime-tokio-rustls", "mysql", "sqlite", "chrono", "uuid"] }
 flate2 = "1.1.5"
 async-compression = { version = "0.4.37", features = ["tokio", "gzip"] }

pg2any-lib/src/config.rs

@@ -514,11 +514,17 @@ impl TableMapping {
 /// Convert from CDC config to replication stream config
 impl From<&Config> for pg_walstream::ReplicationStreamConfig {
     fn from(config: &Config) -> Self {
+        let streaming_mode = if config.streaming {
+            pg_walstream::StreamingMode::On
+        } else {
+            pg_walstream::StreamingMode::Off
+        };
+
         pg_walstream::ReplicationStreamConfig::new(
             config.replication_slot_name.clone(),
             config.publication_name.clone(),
             config.protocol_version,
-            config.streaming,
+            streaming_mode,
             config.heartbeat_interval,
             config.connection_timeout,
             Duration::from_secs(30), // Health check interval

pg2any-lib/src/monitoring/metrics_abstraction.rs

@@ -135,6 +135,7 @@ mod real_metrics {
     use super::*;
     use crate::monitoring::metrics::*; // Import the static metrics
     use crate::types::EventType;
+    use std::borrow::Cow;
     use std::sync::atomic::{AtomicU64, Ordering};
     use std::time::{Duration, Instant};
     use tracing::{debug, warn};
@@ -185,16 +186,22 @@ mod real_metrics {
             // Track by event type and table
             let event_type = event.event_type_str();
             // Extract table name from event type
-            let table_name = match &event.event_type {
+            let table_name: Cow<'_, str> = match &event.event_type {
                 EventType::Insert { table, .. }
                 | EventType::Update { table, .. }
-                | EventType::Delete { table, .. } => table.as_str(),
-                EventType::Truncate(tables) => &tables.join(","),
-                _ => "unknown",
+                | EventType::Delete { table, .. } => Cow::Borrowed(table.as_ref()),
+                EventType::Truncate(tables) => Cow::Owned(
+                    tables
+                        .iter()
+                        .map(|t| t.as_ref())
+                        .collect::<Vec<&str>>()
+                        .join(","),
+                ),
+                _ => Cow::Borrowed("unknown"),
             };
 
             EVENTS_BY_TYPE
-                .with_label_values(&[event_type, table_name])
+                .with_label_values(&[event_type, &table_name])
                 .inc();
 
             // Update LSN

pg2any-lib/src/pg_replication.rs

@@ -70,7 +70,9 @@ impl ReplicationStream {
             .await
             .map_err(|e| crate::error::CdcError::Replication(e))?;
 
-        self.logical_stream.state.update_lsn(event.lsn.value());
+        self.logical_stream
+            .state
+            .update_applied_lsn(event.lsn.value());
 
         Ok(event)
     }