laminar_connectors/fpml/
decoder.rs

1//! The FpML decoder: one owned worker thread per source holding the compiled schema.
2//!
3//! Each document is pre-scanned, parsed and validated by libxml2, reference-checked, and
4//! projected on the worker; the caller blocks for the result. The job queue holds one document,
5//! so a slow validator applies backpressure instead of buffering. Dropping the decoder closes
6//! the queue; the worker finishes the document in hand (native validation cannot be preempted)
7//! and exits, releasing its connector task guard.
8
9use std::sync::mpsc;
10use std::sync::Arc;
11use std::time::Instant;
12
13use arrow_array::RecordBatch;
14use arrow_schema::SchemaRef;
15use sha2::{Digest, Sha256};
16
17use super::artifacts::{hex, Artifact};
18use super::document::Interner;
19use super::error::{Rejection, Stage};
20use super::limits::Limits;
21use super::metrics::{FpmlCounters, FpmlMetrics};
22use super::prescan::prescan;
23use super::profile::{Profile, ProfileSource};
24use super::projection::{project, DocumentInfo};
25use super::references::{ReferencePolicy, ReferenceRules};
26use super::schema_model::SchemaModel;
27use super::xsd::{TreeNames, Validator};
28use crate::config::ConnectorConfig;
29use crate::connector::ConnectorTaskGuard;
30use crate::error::ConnectorError;
31use crate::schema::error::{SchemaError, SchemaResult};
32use crate::schema::traits::FormatDecoder;
33use crate::schema::types::RawRecord;
34
35/// Bundled profile used when neither `fpml.profile` nor `fpml.profile.path` is set: one row per
36/// trade of any asset class, without product economics (the asset-class profiles add those).
37pub const DEFAULT_PROFILE: &str = "fpml-5-13-confirmation-trade";
38/// Names interned beyond the profile's own before the worker resets its interner.
39const MAX_EXTRA_NAMES: usize = 16_384;
40
41/// Parsed `fpml.*` options.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct FpmlConfig {
44    /// Projection profile.
45    pub profile: ProfileSource,
46    /// Directory of an operator-supplied schema closure; the bundled closure otherwise.
47    pub schema_path: Option<String>,
48    /// Reference checking beyond resolution.
49    pub references: ReferencePolicy,
50    /// Per-document bounds.
51    pub limits: Limits,
52}
53
54impl FpmlConfig {
55    /// Parse the `fpml.*` connector options.
56    ///
57    /// # Errors
58    /// Returns a configuration error for conflicting or malformed options.
59    pub fn from_connector_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
60        let profile = match (config.get("fpml.profile"), config.get("fpml.profile.path")) {
61            (Some(_), Some(_)) => {
62                return Err(ConnectorError::ConfigurationError(
63                    "set either 'fpml.profile' or 'fpml.profile.path', not both".into(),
64                ))
65            }
66            (Some(name), None) => ProfileSource::Bundled(name.to_owned()),
67            (None, Some(path)) => ProfileSource::File(path.into()),
68            (None, None) => ProfileSource::Bundled(DEFAULT_PROFILE.to_owned()),
69        };
70        let references = match config.get("fpml.references").unwrap_or("resolve") {
71            "resolve" => ReferencePolicy::Resolve,
72            "schema" => ReferencePolicy::Schema,
73            other => {
74                return Err(ConnectorError::ConfigurationError(format!(
75                    "'fpml.references' must be 'resolve' or 'schema', not '{other}'"
76                )))
77            }
78        };
79        Ok(Self {
80            profile,
81            schema_path: config.get("fpml.schema.path").map(str::to_owned),
82            references,
83            limits: Limits::from_config(config)?,
84        })
85    }
86
87    /// Load the closure and compile the profile. Reads operator files; call off the async
88    /// runtime's worker threads.
89    ///
90    /// # Errors
91    /// Returns a configuration error for an invalid closure or profile.
92    pub fn resolve(&self) -> Result<(Artifact, Profile), ConnectorError> {
93        let artifact = Artifact::resolve(self.schema_path.as_deref())?;
94        let profile = Profile::load(&self.profile, &artifact)?;
95        Ok((artifact, profile))
96    }
97}
98
99struct Job {
100    bytes: Vec<u8>,
101    reply: mpsc::SyncSender<Result<RecordBatch, Rejection>>,
102}
103
104/// Decodes one complete FpML document per record into rows of the profile's schema.
105pub struct FpmlDecoder {
106    schema: SchemaRef,
107    jobs: mpsc::SyncSender<Job>,
108    _metrics: FpmlMetrics,
109}
110
111impl std::fmt::Debug for FpmlDecoder {
112    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
113        f.debug_struct("FpmlDecoder")
114            .field("fields", &self.schema.fields().len())
115            .finish_non_exhaustive()
116    }
117}
118
119impl FpmlDecoder {
120    /// Start the worker: compile the schema on it and wait until it is ready.
121    ///
122    /// `guard` is held by the worker for its lifetime, so the owning connector's task tracker
123    /// observes the worker's exit.
124    ///
125    /// # Errors
126    /// Returns a configuration error if the schema closure does not compile, or an internal
127    /// error if the worker cannot start.
128    pub async fn start(
129        artifact: Artifact,
130        profile: &Profile,
131        config: &FpmlConfig,
132        source: &str,
133        registry: Option<&prometheus::Registry>,
134        guard: Option<ConnectorTaskGuard>,
135    ) -> Result<Self, ConnectorError> {
136        let metrics = match registry {
137            Some(registry) => FpmlMetrics::registered(source, registry)?,
138            None => FpmlMetrics::local(source)?,
139        };
140        let (jobs, queue) = mpsc::sync_channel::<Job>(1);
141        let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
142        let worker = Worker {
143            artifact,
144            plan: Arc::clone(&profile.plan),
145            references: config.references,
146            limits: config.limits,
147            metrics: metrics.counters.clone(),
148        };
149        std::thread::Builder::new()
150            .name("fpml-validator".into())
151            .spawn(move || {
152                let _guard = guard;
153                worker.run(&queue, ready_tx);
154            })
155            .map_err(|error| {
156                ConnectorError::Internal(format!("cannot start the FpML validator thread: {error}"))
157            })?;
158        ready_rx.await.map_err(|_| {
159            ConnectorError::Internal("FpML validator thread exited during startup".into())
160        })??;
161        Ok(Self {
162            schema: profile.schema(),
163            jobs,
164            _metrics: metrics,
165        })
166    }
167
168    /// Decode one document. Blocks the calling thread until the worker answers.
169    ///
170    /// # Errors
171    /// Returns the rejection when any stage rejects the document, or a worker rejection when
172    /// the worker has stopped.
173    pub fn decode(&self, bytes: &[u8]) -> Result<RecordBatch, Rejection> {
174        let (reply, answer) = mpsc::sync_channel(1);
175        self.jobs
176            .send(Job {
177                bytes: bytes.to_vec(),
178                reply,
179            })
180            .map_err(|_| Rejection::new(Stage::Worker, "the FpML validator has stopped"))?;
181        answer.recv().map_err(|_| {
182            Rejection::new(Stage::Worker, "the FpML validator stopped while decoding")
183        })?
184    }
185}
186
187impl FormatDecoder for FpmlDecoder {
188    fn output_schema(&self) -> SchemaRef {
189        Arc::clone(&self.schema)
190    }
191
192    fn decode_batch(&self, records: &[RawRecord]) -> SchemaResult<RecordBatch> {
193        let mut batches = Vec::with_capacity(records.len());
194        for record in records {
195            batches.push(
196                self.decode(&record.value)
197                    .map_err(|rejection| SchemaError::DecodeError(rejection.to_string()))?,
198            );
199        }
200        match batches.len() {
201            0 => Ok(RecordBatch::new_empty(Arc::clone(&self.schema))),
202            1 => Ok(batches.remove(0)),
203            _ => arrow_select::concat::concat_batches(&self.schema, &batches)
204                .map_err(|error| SchemaError::DecodeError(error.to_string())),
205        }
206    }
207
208    fn format_name(&self) -> &'static str {
209        "fpml"
210    }
211}
212
213struct Worker {
214    artifact: Artifact,
215    plan: Arc<super::profile::Plan>,
216    references: ReferencePolicy,
217    limits: Limits,
218    metrics: FpmlCounters,
219}
220
221struct Prepared {
222    validator: Validator,
223    seed: Interner,
224    interner: Interner,
225    names: TreeNames,
226    rules: ReferenceRules,
227}
228
229impl Worker {
230    fn run(
231        self,
232        queue: &mpsc::Receiver<Job>,
233        ready: tokio::sync::oneshot::Sender<Result<(), ConnectorError>>,
234    ) {
235        let mut prepared = match self.prepare() {
236            Ok(prepared) => prepared,
237            Err(error) => {
238                let _ = ready.send(Err(error));
239                return;
240            }
241        };
242        if ready.send(Ok(())).is_err() {
243            return;
244        }
245        // Ends when the decoder (the only sender) is dropped.
246        while let Ok(job) = queue.recv() {
247            let started = Instant::now();
248            self.metrics.document_bytes.inc_by(job.bytes.len() as u64);
249            let result = self.decode(&mut prepared, &job.bytes);
250            self.metrics
251                .decode_seconds
252                .observe(started.elapsed().as_secs_f64());
253            match &result {
254                Ok(batch) => {
255                    self.metrics.documents_accepted.inc();
256                    self.metrics.rows.inc_by(batch.num_rows() as u64);
257                }
258                Err(rejection) => self.metrics.rejected(rejection.stage),
259            }
260            let _ = job.reply.send(result);
261            if prepared.interner.len() > prepared.seed.len() + MAX_EXTRA_NAMES {
262                prepared.interner = prepared.seed.clone();
263            }
264        }
265    }
266
267    fn prepare(&self) -> Result<Prepared, ConnectorError> {
268        let validator = Validator::compile(&self.artifact.files).map_err(|error| {
269            ConnectorError::ConfigurationError(format!(
270                "FpML schema closure '{}': {error}",
271                self.artifact.id
272            ))
273        })?;
274        let model = SchemaModel::for_artifact(&self.artifact)?;
275        let mut interner = self.plan.interner.clone();
276        let names = TreeNames::new(
277            &mut interner,
278            &self.artifact.reference_namespaces,
279            &model.unvalidated_containers(),
280        );
281        let rules = ReferenceRules::new(&model, &mut interner);
282        Ok(Prepared {
283            validator,
284            seed: interner.clone(),
285            interner,
286            names,
287            rules,
288        })
289    }
290
291    fn decode(&self, prepared: &mut Prepared, bytes: &[u8]) -> Result<RecordBatch, Rejection> {
292        let scan = prescan(bytes, &self.limits)?;
293        if scan.root_namespace != self.artifact.namespace {
294            return Err(Rejection::new(
295                Stage::Root,
296                format!(
297                    "the document root is not in the pinned view namespace '{}'",
298                    self.artifact.namespace
299                ),
300            ));
301        }
302        let (document, refusals) = prepared.validator.validate(
303            bytes,
304            &self.limits,
305            &mut prepared.interner,
306            &prepared.names,
307        )?;
308        if refusals.attempts != 0 {
309            return Err(Rejection::new(
310                Stage::Parse,
311                "the document asked libxml2 to load an external resource",
312            ));
313        }
314        if document.elements.len() != scan.elements {
315            return Err(Rejection::new(
316                Stage::Worker,
317                "the pre-scan and the validating parser disagree on the element count",
318            ));
319        }
320        prepared.rules.check(
321            &document,
322            &prepared.interner,
323            prepared.names.href,
324            self.references,
325        )?;
326        let sha256 = hex(&Sha256::digest(bytes));
327        project(
328            &self.plan,
329            &document,
330            &prepared.interner,
331            &prepared.rules,
332            prepared.names.href,
333            &self.limits,
334            &DocumentInfo {
335                sha256: &sha256,
336                bytes: bytes.len(),
337            },
338        )
339    }
340}
laminar_connectors/fpml/decoder.rs

laminar_connectors/fpml/
decoder.rs
