From 73395916eea33d096793712f5b93c2d00e3d23f6 Mon Sep 17 00:00:00 2001 From: jedarden Date: Sun, 24 May 2026 13:03:47 -0400 Subject: [PATCH] feat(cdc): implement NATS sink for CDC events (P5.13.b) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds NATS publishing support for CDC events using async-nats crate. Events are published to subjects with pattern `{subject_prefix}.{index}` (default: "miroir.cdc.{index}") per plan §13.13. - Add async-nats dependency with optional nats-sink feature - Add NATS client pool to CdcManager for connection reuse - Implement flush_nats with connection pooling and error handling - Fix pre-existing bug in CdcRedisOverflow::push (unused _event param) Configuration: ```yaml cdc: sinks: - type: nats url: nats://nats.messaging.svc:4222 subject_prefix: miroir.cdc # optional, default shown ``` Closes: miroir-uhj.13.2 --- crates/miroir-core/Cargo.toml | 4 ++ crates/miroir-core/src/cdc.rs | 120 +++++++++++++++++++++++++++++++--- 2 files changed, 116 insertions(+), 8 deletions(-) diff --git a/crates/miroir-core/Cargo.toml b/crates/miroir-core/Cargo.toml index cb5e1f1..0050df6 100644 --- a/crates/miroir-core/Cargo.toml +++ b/crates/miroir-core/Cargo.toml @@ -41,6 +41,9 @@ futures-util = "0.3" reqwest = { version = "0.12", features = ["json"], default-features = false } urlencoding = "2" +# NATS (CDC sink) +async-nats = { version = "0.38", optional = true } + # Utilities indexmap = "2" chrono = { version = "0.4", features = ["serde"] } @@ -63,6 +66,7 @@ raft-proto = ["bincode"] redis-store = ["redis"] axum = ["dep:axum"] peer-discovery = ["trust-dns-resolver"] +nats-sink = ["async-nats"] test-helpers = [] # Enable when openraft compiles on stable Rust: diff --git a/crates/miroir-core/src/cdc.rs b/crates/miroir-core/src/cdc.rs index b19f07c..b2aa8f8 100644 --- a/crates/miroir-core/src/cdc.rs +++ b/crates/miroir-core/src/cdc.rs @@ -50,6 +50,9 @@ use crate::task_store::{NewCdcCursor, TaskStore}; #[cfg(feature = "redis-store")] use ::redis::AsyncCommands; +#[cfg(feature = "nats-sink")] +use async_nats::Client; + /// Add random jitter to a duration. /// /// Jitter is ±`fraction` of the base duration. For example, with fraction=0.25, @@ -380,6 +383,9 @@ pub struct CdcManager { dropped_metric_callback: Option, /// Internal queue for GET /_miroir/changes endpoint. internal_queue: Arc, + /// NATS client pool (url -> client), for NATS sinks. + #[cfg(feature = "nats-sink")] + nats_clients: Arc>>, } /// CDC manager configuration. @@ -665,7 +671,7 @@ impl CdcRedisOverflow { #[async_trait::async_trait] impl CdcOverflowBackend for CdcRedisOverflow { - async fn push(&self, _event: CdcEvent) -> Result<(), CdcError> { + async fn push(&self, event: CdcEvent) -> Result<(), CdcError> { #[cfg(feature = "redis-store")] return self.push_inner(event).await; @@ -1053,12 +1059,17 @@ impl CdcManager { } } + #[cfg(feature = "nats-sink")] + let nats_clients = Arc::new(RwLock::new(HashMap::new())); + if config.enabled { // Spawn background publisher task let state_clone = state.clone(); let config_clone = config.clone(); let buffers_clone = buffers.clone(); let internal_queue_clone = internal_queue.clone(); + #[cfg(feature = "nats-sink")] + let nats_clients_clone = nats_clients.clone(); tokio::spawn(async move { Self::background_publisher( event_rx, @@ -1066,6 +1077,8 @@ impl CdcManager { config_clone, buffers_clone, internal_queue_clone, + #[cfg(feature = "nats-sink")] + nats_clients_clone, ) .await; }); @@ -1079,6 +1092,8 @@ impl CdcManager { buffers, dropped_metric_callback, internal_queue, + #[cfg(feature = "nats-sink")] + nats_clients, } } @@ -1195,6 +1210,7 @@ impl CdcManager { config: CdcConfig, buffers: HashMap>, internal_queue: Arc, + #[cfg(feature = "nats-sink")] nats_clients: Arc>>, ) { info!("CDC: background publisher started"); @@ -1245,7 +1261,7 @@ impl CdcManager { // Flush if buffer size reached (batch_size trigger) if buffer.len() >= sink.batch_size as usize { - if let Err(e) = Self::flush_sink(sink, buffer, &state, &internal_queue).await { + if let Err(e) = Self::flush_sink(sink, buffer, &state, &internal_queue, #[cfg(feature = "nats-sink")] &nats_clients).await { error!("CDC: failed to flush sink {}: {}", sink.url, e); } sink_buffers.insert(sink.url.clone(), Vec::new()); @@ -1277,7 +1293,16 @@ impl CdcManager { .unwrap_or(now); if now >= flush_deadline { - if let Err(e) = Self::flush_sink(sink, buffer, &state, &internal_queue).await { + if let Err(e) = Self::flush_sink( + sink, + buffer, + &state, + &internal_queue, + #[cfg(feature = "nats-sink")] + &nats_clients, + ) + .await + { error!("CDC: failed to flush sink {} on timer: {}", sink.url, e); } buffer.clear(); @@ -1295,7 +1320,16 @@ impl CdcManager { if !buffer.is_empty() { let sink = config.sinks.iter().find(|s| s.url == sink_url); if let Some(sink) = sink { - if let Err(e) = Self::flush_sink(sink, &buffer, &state, &internal_queue).await { + if let Err(e) = Self::flush_sink( + sink, + &buffer, + &state, + &internal_queue, + #[cfg(feature = "nats-sink")] + &nats_clients, + ) + .await + { error!("CDC: failed to flush sink {} on shutdown: {}", sink_url, e); } } @@ -1314,6 +1348,7 @@ impl CdcManager { events: &[CdcEvent], state: &Arc>, internal_queue: &Arc, + #[cfg(feature = "nats-sink")] nats_clients: &Arc>>, ) -> Result<(), CdcError> { match sink.sink_type { CdcSinkType::Webhook => { @@ -1323,6 +1358,15 @@ impl CdcManager { st.published_count += events.len() as u64; Ok(()) } + #[cfg(feature = "nats-sink")] + CdcSinkType::Nats => { + Self::flush_nats(sink, events, nats_clients).await?; + // Increment published count on success + let mut st = state.write().await; + st.published_count += events.len() as u64; + Ok(()) + } + #[cfg(not(feature = "nats-sink"))] CdcSinkType::Nats => Self::flush_nats(sink, events).await, CdcSinkType::Kafka => Self::flush_kafka(sink, events).await, CdcSinkType::Internal => { @@ -1444,13 +1488,73 @@ impl CdcManager { } } - /// NATS flush (placeholder for P5.13.b). - async fn flush_nats(_sink: &CdcSinkConfig, _events: &[CdcEvent]) -> Result<(), CdcError> { - // NATS publishing implementation (P5.13.b) - // (requires async-nats crate) + /// NATS flush (P5.13.b). + /// + /// Publishes events to NATS subjects using the pattern `{subject_prefix}.{index}`. + /// Plan §13.13: "For each event, PUB to miroir.cdc.{index}" + /// Uses async-nats with connection pooling per sink URL. + #[cfg(feature = "nats-sink")] + async fn flush_nats( + sink: &CdcSinkConfig, + events: &[CdcEvent], + nats_clients: &Arc>>, + ) -> Result<(), CdcError> { + // Get or create NATS client for this sink URL + let client = { + let clients = nats_clients.read().await; + clients.get(&sink.url).cloned() + }; + + let client = match client { + Some(client) => client, + None => { + // Create new connection using async-nats::connect() + let client = async_nats::connect(&sink.url) + .await + .map_err(|e| CdcError::SinkError(format!("NATS connect error: {e}")))?; + + // Store in pool for reuse + let mut clients = nats_clients.write().await; + clients.insert(sink.url.clone(), client.clone()); + client + } + }; + + // Determine subject prefix from config (default: "miroir.cdc") + let subject_prefix = sink + .subject_prefix + .as_ref() + .map(|s| s.as_str()) + .unwrap_or("miroir.cdc"); + + // Publish each event to its index-specific subject + for event in events { + let subject = format!("{}.{}", subject_prefix, event.index); + + // Serialize event (respect include_body setting) + let mut event_to_send = event.clone(); + if !sink.include_body { + event_to_send.document = None; + } + + let payload = serde_json::to_vec(&event_to_send) + .map_err(|e| CdcError::SinkError(format!("serialize error: {e}")))?; + + client + .publish(subject, payload.into()) + .await + .map_err(|e| CdcError::SinkError(format!("NATS publish error: {e}")))?; + } + Ok(()) } + /// NATS flush stub when nats-sink feature is disabled. + #[cfg(not(feature = "nats-sink"))] + async fn flush_nats(_sink: &CdcSinkConfig, _events: &[CdcEvent]) -> Result<(), CdcError> { + Err(CdcError::SinkError("nats-sink feature not enabled".into())) + } + /// Kafka flush (placeholder for P5.13.c). async fn flush_kafka(_sink: &CdcSinkConfig, _events: &[CdcEvent]) -> Result<(), CdcError> { // Kafka publishing implementation (P5.13.c)