camel_component_validator/
xsd_bridge.rs

1use std::sync::Arc;
2use std::sync::atomic::{AtomicUsize, Ordering};
3use std::time::{Duration, Instant};
4
5use async_trait::async_trait;
6use camel_bridge::download::{default_cache_dir_for_spec, ensure_binary_for_spec};
7use camel_bridge::health::wait_for_health;
8use camel_bridge::process::{BridgeError, BridgeProcess, BridgeProcessConfig};
9use camel_bridge::reconnect::BridgeReconnectHandler;
10use camel_bridge::spec::XML_BRIDGE;
11use camel_component_api::RuntimeObservability;
12use dashmap::DashMap;
13use sha2::{Digest, Sha256};
14use std::sync::OnceLock;
15use tokio::sync::{Mutex, RwLock, watch};
16use tonic::Code;
17use tonic::transport::Channel;
18use tracing::error;
19
20use crate::error::ValidatorError;
21use crate::proto;
22use crate::proto::{
23    HealthCheckRequest, RegisterSchemaRequest, RegisterSchemaResponse, ValidateResponse,
24    ValidateWithRequest,
25};
26
27pub type SchemaId = String;
28
29#[derive(Debug, Clone)]
30pub enum BridgeState {
31    Starting,
32    Ready { channel: Channel },
33    Degraded(String),
34    Restarting { attempt: u32, next_at: Instant },
35    Stopped,
36}
37
38pub struct XmlBridgeSlot {
39    pub state_rx: watch::Receiver<BridgeState>,
40    pub(crate) state_tx: watch::Sender<BridgeState>,
41    pub process: Arc<Mutex<Option<BridgeProcess>>>,
42}
43
44impl std::fmt::Debug for XmlBridgeSlot {
45    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46        f.debug_struct("XmlBridgeSlot").finish()
47    }
48}
49
50#[async_trait]
51pub trait XsdBridge: Send + Sync {
52    async fn register(&self, xsd_bytes: Vec<u8>) -> Result<SchemaId, ValidatorError>;
53    async fn validate(&self, schema_id: &str, doc_bytes: Vec<u8>) -> Result<(), ValidatorError>;
54}
55
56#[async_trait]
57trait XsdBridgeRpc: Send + Sync {
58    async fn register_schema(
59        &self,
60        channel: Channel,
61        request: RegisterSchemaRequest,
62    ) -> Result<RegisterSchemaResponse, ValidatorError>;
63
64    async fn validate_with(
65        &self,
66        channel: Channel,
67        request: ValidateWithRequest,
68    ) -> Result<ValidateResponse, ValidatorError>;
69}
70
71#[derive(Debug)]
72struct GrpcXsdBridgeRpc;
73
74#[async_trait]
75impl XsdBridgeRpc for GrpcXsdBridgeRpc {
76    async fn register_schema(
77        &self,
78        channel: Channel,
79        request: RegisterSchemaRequest,
80    ) -> Result<RegisterSchemaResponse, ValidatorError> {
81        let mut client = proto::xsd_validator_client::XsdValidatorClient::new(channel);
82        let response = client.register_schema(request).await.map_err(|e| {
83            ValidatorError::transport_with_source("xml-bridge register_schema RPC failed", e)
84        })?;
85        Ok(response.into_inner())
86    }
87
88    async fn validate_with(
89        &self,
90        channel: Channel,
91        request: ValidateWithRequest,
92    ) -> Result<ValidateResponse, ValidatorError> {
93        let mut client = proto::xsd_validator_client::XsdValidatorClient::new(channel);
94        let response = client.validate_with(request).await.map_err(|e| {
95            ValidatorError::transport_with_source("xml-bridge validate_with RPC failed", e)
96        })?;
97        Ok(response.into_inner())
98    }
99}
100
101#[derive(Clone)]
102pub struct XsdBridgeBackend {
103    channel: Arc<RwLock<Option<Channel>>>,
104    schemas: Arc<DashMap<SchemaId, Vec<u8>>>,
105    schema_cache_max_entries: Arc<AtomicUsize>,
106    slot: Arc<XmlBridgeSlot>,
107    rpc: Arc<dyn XsdBridgeRpc>,
108    start_lock: Arc<Mutex<()>>,
109    bridge_version: String,
110    bridge_cache_dir: std::path::PathBuf,
111    bridge_start_timeout_ms: u64,
112    observability: Arc<OnceLock<(Arc<dyn RuntimeObservability>, String)>>,
113}
114
115impl std::fmt::Debug for XsdBridgeBackend {
116    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
117        f.debug_struct("XsdBridgeBackend")
118            .field("bridge_version", &self.bridge_version)
119            .field("bridge_cache_dir", &self.bridge_cache_dir)
120            .finish()
121    }
122}
123
124impl XsdBridgeBackend {
125    pub fn new() -> Self {
126        let (state_tx, state_rx) = watch::channel(BridgeState::Stopped);
127        let slot = Arc::new(XmlBridgeSlot {
128            state_rx,
129            state_tx,
130            process: Arc::new(Mutex::new(None)),
131        });
132
133        Self {
134            channel: Arc::new(RwLock::new(None)),
135            schemas: Arc::new(DashMap::new()),
136            schema_cache_max_entries: Arc::new(AtomicUsize::new(
137                crate::config::DEFAULT_SCHEMA_CACHE_MAX_ENTRIES,
138            )),
139            slot,
140            rpc: Arc::new(GrpcXsdBridgeRpc),
141            start_lock: Arc::new(Mutex::new(())),
142            bridge_version: crate::BRIDGE_VERSION.to_string(),
143            bridge_cache_dir: default_cache_dir_for_spec(&XML_BRIDGE),
144            bridge_start_timeout_ms: 30_000,
145            observability: Arc::new(OnceLock::new()),
146        }
147    }
148
149    #[cfg(test)]
150    fn for_test(rpc: Arc<dyn XsdBridgeRpc>, channel: Channel) -> Self {
151        let (state_tx, state_rx) = watch::channel(BridgeState::Ready {
152            channel: channel.clone(),
153        });
154        let slot = Arc::new(XmlBridgeSlot {
155            state_rx,
156            state_tx,
157            process: Arc::new(Mutex::new(None)),
158        });
159        Self {
160            channel: Arc::new(RwLock::new(Some(channel))),
161            schemas: Arc::new(DashMap::new()),
162            schema_cache_max_entries: Arc::new(AtomicUsize::new(
163                crate::config::DEFAULT_SCHEMA_CACHE_MAX_ENTRIES,
164            )),
165            slot,
166            rpc,
167            start_lock: Arc::new(Mutex::new(())),
168            bridge_version: crate::BRIDGE_VERSION.to_string(),
169            bridge_cache_dir: default_cache_dir_for_spec(&XML_BRIDGE),
170            bridge_start_timeout_ms: 30_000,
171            observability: Arc::new(OnceLock::new()),
172        }
173    }
174
175    pub fn set_observability(&self, runtime: Arc<dyn RuntimeObservability>, route_id: String) {
176        self.observability.set((runtime, route_id)).ok();
177    }
178
179    pub fn schema_id_for(xsd_bytes: &[u8]) -> SchemaId {
180        let mut hasher = Sha256::new();
181        hasher.update(xsd_bytes);
182        format!("xsd-{}", hex::encode(hasher.finalize()))
183    }
184
185    /// Update the maximum number of entries allowed in the schema cache.
186    /// If the cache already exceeds the new limit, it is cleared immediately.
187    pub fn set_schema_cache_max_entries(&self, max_entries: usize) {
188        if self.schemas.len() > max_entries {
189            self.schemas.clear();
190        }
191        self.schema_cache_max_entries
192            .store(max_entries, Ordering::Relaxed);
193    }
194
195    async fn ensure_bridge_ready(&self) -> Result<Channel, ValidatorError> {
196        if let Some(ch) = self.channel.read().await.clone() {
197            return Ok(ch);
198        }
199
200        let _guard = self.start_lock.lock().await;
201        if let Some(ch) = self.channel.read().await.clone() {
202            return Ok(ch);
203        }
204
205        let _ = self.slot.state_tx.send(BridgeState::Starting);
206        let (process, channel) = self.start_bridge_process().await?;
207        {
208            let mut process_guard = self.slot.process.lock().await;
209            *process_guard = Some(process);
210        }
211        {
212            let mut ch_guard = self.channel.write().await;
213            *ch_guard = Some(channel.clone());
214        }
215        let _ = self.slot.state_tx.send(BridgeState::Ready {
216            channel: channel.clone(),
217        });
218        self.on_reconnect(&channel).map_err(|e| {
219            ValidatorError::transport_with_source("xml-bridge reconnect handler failed", e)
220        })?;
221
222        Ok(channel)
223    }
224
225    async fn restart_bridge(&self) -> Result<Channel, ValidatorError> {
226        let _guard = self.start_lock.lock().await;
227
228        let _ = self.slot.state_tx.send(BridgeState::Restarting {
229            attempt: 0,
230            next_at: Instant::now(),
231        });
232
233        let old_process = {
234            let mut process_guard = self.slot.process.lock().await;
235            process_guard.take()
236        };
237        if let Some(p) = old_process {
238            let _ = p.stop().await;
239        }
240
241        let (process, channel) = self.start_bridge_process().await?;
242        {
243            let mut process_guard = self.slot.process.lock().await;
244            *process_guard = Some(process);
245        }
246        {
247            let mut ch_guard = self.channel.write().await;
248            *ch_guard = Some(channel.clone());
249        }
250
251        self.on_reconnect(&channel).map_err(|e| {
252            ValidatorError::transport_with_source("xml-bridge reconnect handler failed", e)
253        })?;
254        let _ = self.slot.state_tx.send(BridgeState::Ready {
255            channel: channel.clone(),
256        });
257
258        Ok(channel)
259    }
260
261    async fn start_bridge_process(&self) -> Result<(BridgeProcess, Channel), ValidatorError> {
262        let binary_path =
263            ensure_binary_for_spec(&XML_BRIDGE, &self.bridge_version, &self.bridge_cache_dir)
264                .await
265                .map_err(|e| {
266                    ValidatorError::endpoint(format!("XML bridge binary unavailable: {e}"))
267                })?;
268
269        let process_config = BridgeProcessConfig::xml(binary_path, self.bridge_start_timeout_ms);
270        let (process, channel) = BridgeProcess::start_and_connect(&process_config)
271            .await
272            .map_err(|e| ValidatorError::endpoint(format!("XML bridge start failed: {e}")))?;
273
274        wait_for_health(&channel, Duration::from_secs(10), |ch| {
275            let mut client = proto::health_client::HealthClient::new(ch);
276            async move {
277                let resp = client.check(HealthCheckRequest {}).await?;
278                Ok(resp.into_inner().status == "SERVING")
279            }
280        })
281        .await
282        .map_err(|e| ValidatorError::endpoint(format!("XML bridge health check failed: {e}")))?;
283
284        Ok((process, channel))
285    }
286
287    fn is_transport_error(msg: &str) -> bool {
288        msg.contains(&Code::Unavailable.to_string())
289            || msg.contains(&Code::Unknown.to_string())
290            || msg.contains("transport")
291    }
292
293    pub async fn shutdown(&self) {
294        let mut guard = self.slot.process.lock().await;
295        if let Some(p) = guard.take()
296            && let Err(e) = p.stop().await
297        {
298            tracing::warn!("Failed to stop XSD bridge process: {}", e);
299        }
300    }
301
302    async fn register_with_channel(
303        &self,
304        channel: Channel,
305        schema_id: SchemaId,
306        xsd_bytes: Vec<u8>,
307    ) -> Result<SchemaId, ValidatorError> {
308        let response = self
309            .rpc
310            .register_schema(
311                channel,
312                RegisterSchemaRequest {
313                    schema_id: schema_id.clone(),
314                    schema: xsd_bytes.clone(),
315                },
316            )
317            .await?;
318
319        if let Some(err) = response.error {
320            return Err(ValidatorError::from_bridge_error(&err));
321        }
322
323        // Evict the entire cache if it has reached capacity. A full clear is
324        // acceptable because schemas are re-registered on demand and also
325        // re-seeded after bridge reconnects.
326        let max_entries = self.schema_cache_max_entries.load(Ordering::Relaxed);
327        if self.schemas.len() >= max_entries && !self.schemas.contains_key(&schema_id) {
328            self.schemas.clear();
329        }
330
331        self.schemas.insert(schema_id.clone(), xsd_bytes);
332        Ok(schema_id)
333    }
334}
335
336impl BridgeReconnectHandler for XsdBridgeBackend {
337    fn on_reconnect(&self, _channel: &Channel) -> Result<(), BridgeError> {
338        let this = self.clone();
339        tokio::spawn(async move {
340            let Some(channel) = this.channel.read().await.clone() else {
341                return;
342            };
343
344            let schemas: Vec<(SchemaId, Vec<u8>)> = this
345                .schemas
346                .iter()
347                .map(|entry| (entry.key().clone(), entry.value().clone()))
348                .collect();
349
350            let observability = this
351                .observability
352                .get()
353                .map(|(rt, rid)| (Arc::clone(rt), rid.clone()));
354
355            for (schema_id, schema_bytes) in schemas {
356                if let Err(e) = this
357                    .rpc
358                    .register_schema(
359                        channel.clone(),
360                        RegisterSchemaRequest {
361                            schema_id: schema_id.clone(),
362                            schema: schema_bytes,
363                        },
364                    )
365                    .await
366                {
367                    if let Some((ref rt, ref rid)) = observability {
368                        rt.metrics()
369                            .increment_errors(rid, "e:validator:reconnect-reseed");
370                    }
371                    // log-policy: outside-contract
372                    error!(schema_id = %schema_id, error = %e, "re-seed schema failed after reconnect");
373                }
374            }
375        });
376        Ok(())
377    }
378}
379
380#[async_trait]
381impl XsdBridge for XsdBridgeBackend {
382    async fn register(&self, xsd_bytes: Vec<u8>) -> Result<SchemaId, ValidatorError> {
383        let schema_id = Self::schema_id_for(&xsd_bytes);
384        if self.schemas.contains_key(&schema_id) {
385            return Ok(schema_id);
386        }
387
388        let channel = self.ensure_bridge_ready().await?;
389        match self
390            .register_with_channel(channel.clone(), schema_id.clone(), xsd_bytes.clone())
391            .await
392        {
393            Ok(id) => Ok(id),
394            Err(e) if Self::is_transport_error(&e.to_string()) => {
395                let restarted = self.restart_bridge().await?;
396                self.register_with_channel(restarted, schema_id, xsd_bytes)
397                    .await
398            }
399            Err(e) => Err(e),
400        }
401    }
402
403    async fn validate(&self, schema_id: &str, doc_bytes: Vec<u8>) -> Result<(), ValidatorError> {
404        let channel = self.ensure_bridge_ready().await?;
405        let req = ValidateWithRequest {
406            schema_id: schema_id.to_string(),
407            document: doc_bytes.clone(),
408        };
409
410        let response = match self.rpc.validate_with(channel.clone(), req).await {
411            Ok(resp) => resp,
412            Err(e) if Self::is_transport_error(&e.to_string()) => {
413                let restarted = self.restart_bridge().await?;
414                self.rpc
415                    .validate_with(
416                        restarted,
417                        ValidateWithRequest {
418                            schema_id: schema_id.to_string(),
419                            document: doc_bytes,
420                        },
421                    )
422                    .await?
423            }
424            Err(e) => return Err(e),
425        };
426
427        if let Some(err) = response.error {
428            return Err(ValidatorError::from_bridge_error(&err));
429        }
430        if response.valid {
431            return Ok(());
432        }
433
434        let details = response
435            .errors
436            .iter()
437            .map(|e| format!("{}:{} {}", e.line, e.column, e.message))
438            .collect::<Vec<_>>()
439            .join("\n");
440        Err(ValidatorError::validation(format!(
441            "XSD validation failed:\n{details}"
442        )))
443    }
444}
445
446impl Default for XsdBridgeBackend {
447    fn default() -> Self {
448        Self::new()
449    }
450}
451
452#[cfg(test)]
453mod tests {
454    use super::*;
455    use std::sync::atomic::{AtomicUsize, Ordering};
456    use tonic::transport::Endpoint;
457
458    #[derive(Debug)]
459    struct MockRpc {
460        register_calls: Arc<AtomicUsize>,
461        validate_ok: bool,
462    }
463
464    #[async_trait]
465    impl XsdBridgeRpc for MockRpc {
466        async fn register_schema(
467            &self,
468            _channel: Channel,
469            request: RegisterSchemaRequest,
470        ) -> Result<RegisterSchemaResponse, ValidatorError> {
471            self.register_calls.fetch_add(1, Ordering::SeqCst);
472            Ok(RegisterSchemaResponse {
473                schema_id: request.schema_id,
474                error: None,
475            })
476        }
477
478        async fn validate_with(
479            &self,
480            _channel: Channel,
481            _request: ValidateWithRequest,
482        ) -> Result<ValidateResponse, ValidatorError> {
483            Ok(ValidateResponse {
484                valid: self.validate_ok,
485                errors: Vec::new(),
486                error: None,
487            })
488        }
489    }
490
491    fn lazy_channel() -> Channel {
492        Endpoint::from_static("http://127.0.0.1:65535").connect_lazy()
493    }
494
495    #[tokio::test]
496    async fn xsd_bridge_reconnect_reseeds() {
497        let calls = Arc::new(AtomicUsize::new(0));
498        let rpc = Arc::new(MockRpc {
499            register_calls: Arc::clone(&calls),
500            validate_ok: true,
501        });
502        let channel = lazy_channel();
503        let backend = XsdBridgeBackend::for_test(rpc, channel);
504
505        let _id_a = backend.register(b"<xsd:a/>".to_vec()).await.unwrap();
506        let _id_b = backend.register(b"<xsd:b/>".to_vec()).await.unwrap();
507
508        backend.on_reconnect(&lazy_channel()).unwrap();
509        tokio::time::sleep(Duration::from_millis(20)).await;
510
511        assert!(calls.load(Ordering::SeqCst) >= 4);
512    }
513}
camel_component_validator/xsd_bridge.rs

camel_component_validator/
xsd_bridge.rs
