laminar_connectors/fix/
source.rs

1//! `SourceConnector` implementation: one FIX initiator session per source.
2
3use std::sync::Arc;
4use std::time::Duration;
5
6use arrow_schema::{Schema, SchemaRef};
7use async_trait::async_trait;
8use tokio::sync::Notify;
9
10use super::capture::{Capture, SessionStatus};
11use super::config::{FixSourceConfig, SequenceResetPolicy, Transport};
12use super::dictionary::DictionaryDigests;
13use super::engine::{Application, Initiator, Settings, StoreFactory};
14use super::journal::{GroupCommit, Journal, JournalReader, Record};
15use super::metrics::{saturating, FixMetrics};
16use super::projection::{compile, Projection};
17use super::session_store::SessionStore;
18use super::tls::TlsBridge;
19use super::venue::Logon;
20use super::{dictionary, NATIVE_FORMAT};
21use crate::checkpoint::SourceCheckpoint;
22use crate::config::ConnectorConfig;
23use crate::connector::{
24    ConnectorTaskGuard, ConnectorTaskOwner, ConnectorTaskTracker, SourceBatch, SourceConnector,
25    SourceConsistency, SourceContract, SourceInputMode, SourcePosition, SourceReplayOrder,
26    SourceRowPositionCapability, SourceStart, SourceTopology,
27};
28use crate::error::ConnectorError;
29use crate::schema::resolution::{
30    bind_external, logical_binding, NativeSchema, SchemaBinding, SchemaDirection, SchemaOrigin,
31};
32
33const CHECKPOINT_CONNECTOR: &str = "fix";
34const CHECKPOINT_VERSION: &str = "1";
35const OFFSET_NEXT: &str = "journal.next";
36/// Upper bound on journal bytes projected by one poll, independent of `max_records`.
37const MAX_POLL_BYTES: usize = 8 * 1024 * 1024;
38/// Bound on waiting for the engine's graceful Logout (the engine itself waits at most ten
39/// seconds for the counterparty's response).
40const GRACEFUL_STOP: Duration = Duration::from_secs(12);
41
42/// FIX initiator source: captures selected application messages into a durable journal and
43/// emits them as Arrow rows.
44pub struct FixSource {
45    registry: Option<Arc<prometheus::Registry>>,
46    schema: SchemaRef,
47    data_ready: Arc<Notify>,
48    task_owner: ConnectorTaskOwner,
49    task_tracker: ConnectorTaskTracker,
50    running: Option<Running>,
51}
52
53struct Engine {
54    initiator: Option<Initiator>,
55    tls: Option<TlsBridge>,
56    guard: Option<ConnectorTaskGuard>,
57}
58
59impl Engine {
60    /// Stop and delete the initiator on a dedicated thread that owns the task guard, so joining
61    /// engine threads never blocks an async worker and the journal lock is released only after
62    /// the last callback returned.
63    fn spawn_stop(&mut self, force: bool) -> Option<tokio::sync::oneshot::Receiver<()>> {
64        let initiator = self.initiator.take()?;
65        let guard = self.guard.take();
66        let (done, finished) = tokio::sync::oneshot::channel();
67        let stop: Box<dyn FnOnce() + Send> = Box::new(move || {
68            if let Err(error) = initiator.stop(force) {
69                tracing::warn!(error = %error, "FIX engine stop failed");
70            }
71            drop(initiator);
72            drop(guard);
73            let _ = done.send(());
74        });
75        let stop = Arc::new(parking_lot::Mutex::new(Some(stop)));
76        let threaded = stop.clone();
77        let spawned = std::thread::Builder::new()
78            .name("laminardb-fix-engine-stop".into())
79            .spawn(move || {
80                let stop = threaded.lock().take();
81                if let Some(stop) = stop {
82                    stop();
83                }
84            });
85        if let Err(error) = spawned {
86            tracing::error!(error = %error, "cannot spawn the FIX engine stop thread; stopping inline");
87            let stop = stop.lock().take();
88            if let Some(stop) = stop {
89                stop();
90            }
91        }
92        Some(finished)
93    }
94
95    /// Log out gracefully within `GRACEFUL_STOP`, then close the TLS bridge.
96    async fn stop(mut self) {
97        if let Some(finished) = self.spawn_stop(false) {
98            if tokio::time::timeout(GRACEFUL_STOP, finished).await.is_err() {
99                tracing::warn!(
100                    "FIX engine logout did not finish in time; its stop thread continues"
101                );
102            }
103        }
104        if let Some(tls) = self.tls.take() {
105            tls.stop().await;
106        }
107    }
108}
109
110impl Drop for Engine {
111    fn drop(&mut self) {
112        let _ = self.spawn_stop(true);
113    }
114}
115
116struct Running {
117    journal: Arc<Journal>,
118    store: Arc<SessionStore>,
119    _group_commit: Option<GroupCommit>,
120    /// Cursor metadata and input-channel inventory shared by every checkpoint of this run.
121    cursor: SourceCheckpoint,
122    reader: JournalReader,
123    projection: Projection,
124    status: Arc<SessionStatus>,
125    metrics: Arc<FixMetrics>,
126    on_sequence_reset: SequenceResetPolicy,
127    /// Whether the counterparty answers ResendRequest with the original messages.
128    resends: bool,
129    /// Journal index of the latest committed checkpoint.
130    committed: u64,
131    engine: Option<Engine>,
132    #[cfg(feature = "fpml")]
133    fpml: Option<super::fpml::Documents>,
134}
135
136impl Running {
137    /// Discard partially projected rows after an error.
138    fn discard(&mut self) {
139        self.projection.reset();
140        #[cfg(feature = "fpml")]
141        if let Some(fpml) = &mut self.fpml {
142            fpml.reset();
143        }
144    }
145}
146
147impl FixSource {
148    /// Create an unstarted source. Construction performs no I/O.
149    #[must_use]
150    pub fn new(registry: Option<Arc<prometheus::Registry>>) -> Self {
151        let (task_owner, task_tracker) = ConnectorTaskOwner::new();
152        Self {
153            registry,
154            schema: Arc::new(Schema::empty()),
155            data_ready: Arc::new(Notify::new()),
156            task_owner,
157            task_tracker,
158            running: None,
159        }
160    }
161
162    fn guard(&self) -> Result<ConnectorTaskGuard, ConnectorError> {
163        self.task_owner
164            .track()
165            .ok_or_else(|| ConnectorError::Internal("FIX source generation is retired".into()))
166    }
167
168    /// The session is the source's one input channel, matching the row-position partition.
169    fn cursor_template(journal: &Journal) -> Result<SourceCheckpoint, ConnectorError> {
170        let mut checkpoint = SourceCheckpoint::new();
171        checkpoint.set_metadata("connector", CHECKPOINT_CONNECTOR);
172        checkpoint.set_metadata("checkpoint.version", CHECKPOINT_VERSION);
173        checkpoint.set_metadata("journal.id", journal.meta().journal_id.clone());
174        checkpoint.set_metadata("capture.contract", journal.meta().capture_contract.clone());
175        checkpoint.set_input_channels(vec![journal.meta().session.as_bytes().to_vec()])?;
176        Ok(checkpoint)
177    }
178
179    fn checkpoint_for(template: &SourceCheckpoint, next: u64) -> SourceCheckpoint {
180        let mut checkpoint = template.clone();
181        checkpoint.set_offset(OFFSET_NEXT, next.to_string());
182        checkpoint
183    }
184
185    fn checkpoint_index(
186        journal: &Journal,
187        checkpoint: &SourceCheckpoint,
188    ) -> Result<u64, ConnectorError> {
189        let meta = journal.meta();
190        let matches = checkpoint.get_metadata("connector") == Some(CHECKPOINT_CONNECTOR)
191            && checkpoint.get_metadata("checkpoint.version") == Some(CHECKPOINT_VERSION)
192            && checkpoint.get_metadata("journal.id") == Some(meta.journal_id.as_str())
193            && checkpoint.get_metadata("capture.contract") == Some(meta.capture_contract.as_str());
194        if !matches {
195            return Err(ConnectorError::ConfigurationError(
196                "FIX source checkpoint belongs to a different journal or capture contract; the \
197                 journal was replaced or the profile changed incompatibly"
198                    .into(),
199            ));
200        }
201        checkpoint
202            .get_offset(OFFSET_NEXT)
203            .and_then(|value| value.parse().ok())
204            .ok_or_else(|| {
205                ConnectorError::ConfigurationError(
206                    "FIX source checkpoint has no journal position".into(),
207                )
208            })
209    }
210
211    fn running(&mut self) -> Result<&mut Running, ConnectorError> {
212        self.running
213            .as_mut()
214            .ok_or_else(|| ConnectorError::InvalidState {
215                expected: "started".into(),
216                actual: "not started".into(),
217            })
218    }
219}
220
221fn committed_digests(binding: Option<&SchemaBinding>) -> Option<DictionaryDigests> {
222    let native = binding?.value.as_ref()?;
223    if native.format != NATIVE_FORMAT {
224        return None;
225    }
226    Some(DictionaryDigests {
227        application: native.identity.get("dictionary_sha256")?.clone(),
228        transport: native.identity.get("transport_dictionary_sha256").cloned(),
229    })
230}
231
232/// Open the journal and the session store whose persisted inbound progress it bounds.
233fn open_durable_state(
234    parsed: &FixSourceConfig,
235    digests: &DictionaryDigests,
236    data_ready: Arc<Notify>,
237) -> Result<(Arc<Journal>, Arc<SessionStore>), ConnectorError> {
238    let journal = Arc::new(Journal::open(
239        &parsed.journal_path,
240        &parsed.session.display(),
241        &parsed.capture_contract(digests),
242        parsed.journal,
243        data_ready,
244    )?);
245    let barrier_journal = Arc::downgrade(&journal);
246    let store = Arc::new(SessionStore::open(
247        &parsed.journal_path,
248        parsed.journal.sync,
249        Box::new(move || match barrier_journal.upgrade() {
250            Some(journal) => journal.sync(),
251            None => Err(terminal("FIX journal closed".into())),
252        }),
253    )?);
254    journal.attach_floor(store.clone())?;
255    Ok((journal, store))
256}
257
258fn terminal(reason: String) -> ConnectorError {
259    ConnectorError::InvalidState {
260        expected: "FIX capture running".into(),
261        actual: reason,
262    }
263}
264
265#[async_trait]
266impl SourceConnector for FixSource {
267    fn terminal_task_tracker(&self) -> Option<ConnectorTaskTracker> {
268        Some(self.task_tracker.clone())
269    }
270
271    fn contract(&self, config: &ConnectorConfig) -> Result<SourceContract, ConnectorError> {
272        FixSourceConfig::parse(config)?;
273        // Journal retention advances only on committed checkpoints. One session is one ordered
274        // channel; cluster placement is rejected because the journal is node-local. The barrier
275        // cut is exact: a cursor is a journal index, every emitted record is durable, and one
276        // FIX message is one atomic record with no upstream transaction spanning the cut.
277        Ok(SourceContract::new(
278            SourceConsistency::CommitCoupled,
279            SourceTopology::Singleton,
280            SourceInputMode::AppendOnly,
281        )
282        .with_row_positions(SourceRowPositionCapability::OrderedDeterministic)
283        .with_replay_order(SourceReplayOrder::SingleChannel)
284        .with_exact_delivery_certification())
285    }
286
287    fn recovery_identity_options(
288        &self,
289        config: &ConnectorConfig,
290    ) -> Result<Option<std::collections::BTreeMap<String, String>>, ConnectorError> {
291        Ok(Some(FixSourceConfig::parse(config)?.recovery_identity()))
292    }
293
294    async fn resolve_schema(
295        &mut self,
296        config: &ConnectorConfig,
297        explicit: Option<SchemaRef>,
298    ) -> Result<SchemaBinding, ConnectorError> {
299        let schema = explicit.ok_or_else(|| {
300            ConnectorError::ConfigurationError(
301                "FIX sources need declared columns mapped through 'column.<name>' options".into(),
302            )
303        })?;
304        let parsed = FixSourceConfig::parse(config)?;
305        let files = parsed.dictionary_files();
306        let digests = dictionary::digests(&files)?;
307        let dictionaries = dictionary::load(&parsed, &files)?;
308        #[cfg(feature = "fpml")]
309        let fpml = match super::fpml::field_tag(config)? {
310            Some(tag) => Some(
311                super::fpml::resolve(config, tag, &schema, &dictionaries, &parsed.messages).await?,
312            ),
313            None => None,
314        };
315        let plan = compile(config, &schema, &dictionaries, &parsed.messages)?;
316        #[cfg_attr(not(feature = "fpml"), allow(unused_mut))]
317        let mut output = Arc::clone(&schema);
318        let mut identity = std::collections::BTreeMap::from([
319            ("dictionary_sha256".into(), digests.application),
320            ("session".into(), parsed.session.display()),
321        ]);
322        #[cfg(feature = "fpml")]
323        if let Some(resolved) = fpml {
324            output = resolved.output;
325            identity.extend(resolved.identity);
326        }
327        let mut binding = logical_binding(
328            config,
329            SchemaDirection::Source,
330            SchemaOrigin::Explicit,
331            &output,
332        )?;
333        bind_external(&mut binding, &output)?;
334        if let Some(transport) = digests.transport {
335            identity.insert("transport_dictionary_sha256".into(), transport);
336        }
337        binding.value = Some(NativeSchema {
338            format: NATIVE_FORMAT.into(),
339            identity,
340            definition: plan.canonical(),
341            references: Vec::new(),
342        });
343        Ok(binding)
344    }
345
346    #[allow(clippy::too_many_lines)] // INVARIANT: one ordered activation transaction: pin, open, position, start.
347    async fn start(&mut self, request: SourceStart) -> Result<(), ConnectorError> {
348        let (config, position, _delivery) = request.into_parts();
349        let parsed = FixSourceConfig::parse(&config)?;
350        let schema = config.arrow_schema().ok_or_else(|| {
351            ConnectorError::ConfigurationError(
352                "FIX source started without its declared schema".into(),
353            )
354        })?;
355        let files = parsed.dictionary_files();
356        let digests = match committed_digests(config.schema_binding()) {
357            Some(digests) => digests,
358            None => dictionary::digests(&files)?,
359        };
360        let pinned = dictionary::pin_all(&parsed.journal_path, &digests, &files)?;
361        let dictionaries = Arc::new(dictionary::load(&parsed, &pinned)?);
362        #[cfg(feature = "fpml")]
363        let (schema, fpml) = super::fpml::open(
364            &config,
365            schema,
366            &dictionaries,
367            self.registry.as_deref(),
368            self.task_owner.admission(),
369        )
370        .await?;
371        let plan = compile(&config, &schema, &dictionaries, &parsed.messages)?;
372        #[cfg(feature = "fpml")]
373        let schema = fpml.as_ref().map_or(schema, super::fpml::Documents::schema);
374        let logon = Logon::load(&parsed)?;
375        let session = parsed.session.display();
376        let (journal, store) = open_durable_state(&parsed, &digests, self.data_ready.clone())?;
377        let start_index = match &position {
378            SourcePosition::Initial => journal.first_retained_index(),
379            SourcePosition::Resume { checkpoint, .. } => {
380                Self::checkpoint_index(&journal, checkpoint)?
381            }
382            SourcePosition::Initialized { .. } => {
383                return Err(ConnectorError::FeatureUnsupported(
384                    "FIX sources have no sealed topology initialization contract".into(),
385                ))
386            }
387        };
388        let reader = journal.reader(start_index)?;
389        let replay = journal.durable_end().next_index.saturating_sub(start_index);
390        let cursor = Self::cursor_template(&journal)?;
391        let source_name = config.get("laminar.source.name").unwrap_or("fix");
392        let metrics = Arc::new(match &self.registry {
393            Some(registry) => FixMetrics::registered(source_name, registry)?,
394            None => FixMetrics::local(source_name)?,
395        });
396        metrics
397            .journal_retained_bytes
398            .set(saturating(journal.retained_bytes()));
399        metrics.startup_replay_records.set(saturating(replay));
400        let status = Arc::new(SessionStatus::default());
401
402        let tls = match &parsed.transport {
403            Transport::Tcp => None,
404            Transport::Tls(settings) => {
405                Some(TlsBridge::start(settings, &parsed.host, parsed.port, self.guard()?).await?)
406            }
407        };
408        let connect = match &tls {
409            Some(bridge) => ("127.0.0.1".to_string(), bridge.local_port),
410            None => (parsed.host.clone(), parsed.port),
411        };
412        let defaults = parsed.engine_defaults(&pinned, (&connect.0, connect.1));
413        let settings = Settings::single_session(&defaults, &parsed.session, &[])?;
414        let capture = Arc::new(Capture::new(
415            journal.clone(),
416            store.clone(),
417            &parsed,
418            logon,
419            status.clone(),
420            metrics.clone(),
421        ));
422        let guard = self.guard()?;
423        let group_commit = journal.start_group_commit()?;
424        let initiator = Initiator::new(
425            settings,
426            Application::new(capture)?,
427            StoreFactory::owned(store.clone())?,
428        )?;
429        initiator.start()?;
430        let engine = Engine {
431            initiator: Some(initiator),
432            tls,
433            guard: Some(guard),
434        };
435        tracing::info!(
436            session = %session,
437            start_index,
438            durability = ?journal.sync_policy(),
439            "FIX source started"
440        );
441        self.schema = schema;
442        self.running = Some(Running {
443            projection: Projection::new(
444                plan,
445                dictionaries,
446                Arc::from(session),
447                parsed.max_rows_per_message,
448            ),
449            journal,
450            store,
451            _group_commit: group_commit,
452            cursor,
453            reader,
454            status,
455            metrics,
456            on_sequence_reset: parsed.on_sequence_reset,
457            resends: parsed.resends,
458            committed: start_index,
459            engine: Some(engine),
460            #[cfg(feature = "fpml")]
461            fpml,
462        });
463        self.data_ready.notify_one();
464        Ok(())
465    }
466
467    async fn poll_batch(
468        &mut self,
469        max_records: usize,
470    ) -> Result<Option<SourceBatch>, ConnectorError> {
471        let data_ready = self.data_ready.clone();
472        let running = self.running()?;
473        let stopped = running
474            .status
475            .fatal()
476            .or_else(|| running.journal.poisoned())
477            .or_else(|| running.store.poisoned());
478        if let Some(reason) = stopped {
479            return Err(terminal(reason));
480        }
481        let end = running.journal.durable_end();
482        let mut bytes = 0_usize;
483        while running.projection.rows() < max_records.max(1) && bytes < MAX_POLL_BYTES {
484            let Some((index, record)) = running.reader.next(&end)? else {
485                break;
486            };
487            match record {
488                Record::Application(application) => {
489                    bytes += application.message.len();
490                    #[cfg_attr(not(feature = "fpml"), allow(unused_variables))]
491                    let rows = match running.projection.push(index, &application) {
492                        Ok(rows) => rows,
493                        Err(error) => {
494                            running.discard();
495                            running.metrics.projection_errors.inc();
496                            return Err(error.into());
497                        }
498                    };
499                    #[cfg(feature = "fpml")]
500                    if let Some(fpml) = &mut running.fpml {
501                        if let Err(error) = fpml.capture(index, &application, rows) {
502                            running.discard();
503                            running.metrics.projection_errors.inc();
504                            return Err(error);
505                        }
506                    }
507                }
508                Record::SequenceReset {
509                    from_seq, new_seq, ..
510                } => {
511                    running.metrics.feed_incomplete.set(1);
512                    match running.on_sequence_reset {
513                        SequenceResetPolicy::Fail => {
514                            running.discard();
515                            return Err(terminal(format!(
516                                "counterparty skipped inbound sequence numbers {from_seq}..{new_seq} with \
517                                 SequenceReset-Reset; that history is unrecoverable. Reconcile it, then set \
518                                 'on.sequence.reset' = 'continue' to proceed past journal record {index}"
519                            )));
520                        }
521                        SequenceResetPolicy::Continue => running.projection.mark_gap(),
522                    }
523                }
524                // Without resends, a new series or a gap fill may stand in for lost messages.
525                Record::SeriesStarted { series: 2.., .. } | Record::GapFill { .. }
526                    if !running.resends =>
527                {
528                    running.metrics.feed_incomplete.set(1);
529                    running.projection.mark_gap();
530                }
531                Record::SessionSnapshot(_)
532                | Record::SeriesStarted { .. }
533                | Record::GapFill { .. }
534                | Record::Rejected { .. } => {}
535            }
536        }
537        let unread = end.next_index.saturating_sub(running.reader.next_index());
538        running
539            .metrics
540            .checkpoint_lag_records
541            .set(saturating(end.next_index.saturating_sub(running.committed)));
542        running
543            .metrics
544            .journal_unread_records
545            .set(saturating(unread));
546        if unread > 0 {
547            data_ready.notify_one();
548        }
549        if running.projection.rows() == 0 {
550            return Ok(None);
551        }
552        let (records, positions) = running.projection.finish()?;
553        #[cfg(feature = "fpml")]
554        let records = match &mut running.fpml {
555            Some(fpml) => match fpml.append(records).await {
556                Ok(records) => records,
557                Err(error) => {
558                    running.metrics.projection_errors.inc();
559                    return Err(error);
560                }
561            },
562            None => records,
563        };
564        running
565            .metrics
566            .rows_emitted
567            .inc_by(records.num_rows() as u64);
568        let checkpoint = Self::checkpoint_for(&running.cursor, running.reader.next_index());
569        Ok(Some(
570            SourceBatch::positioned(records, positions)?.with_checkpoint(checkpoint),
571        ))
572    }
573
574    fn schema(&self) -> SchemaRef {
575        self.schema.clone()
576    }
577
578    fn checkpoint(&self) -> SourceCheckpoint {
579        match &self.running {
580            Some(running) => Self::checkpoint_for(&running.cursor, running.reader.next_index()),
581            None => SourceCheckpoint::new(),
582        }
583    }
584
585    async fn close(&mut self) -> Result<(), ConnectorError> {
586        if let Some(mut running) = self.running.take() {
587            if let Some(engine) = running.engine.take() {
588                engine.stop().await;
589            }
590            // Lets the session store persist inbound progress up to the last record.
591            if let Err(error) = running.journal.sync() {
592                tracing::warn!(error = %error, "FIX journal sync on close failed");
593            }
594            tracing::info!(
595                session = %running.journal.meta().session,
596                next = running.reader.next_index(),
597                "FIX source closed"
598            );
599        }
600        Ok(())
601    }
602
603    fn data_ready_notify(&self) -> Option<Arc<Notify>> {
604        Some(self.data_ready.clone())
605    }
606
607    async fn notify_epoch_committed(
608        &mut self,
609        _epoch: u64,
610        checkpoint: &SourceCheckpoint,
611    ) -> Result<(), ConnectorError> {
612        if checkpoint.is_empty() {
613            return Ok(());
614        }
615        let running = self.running()?;
616        let committed = Self::checkpoint_index(&running.journal, checkpoint)?;
617        running.journal.release_before(committed)?;
618        running.committed = running.committed.max(committed);
619        running
620            .metrics
621            .journal_retained_bytes
622            .set(saturating(running.journal.retained_bytes()));
623        Ok(())
624    }
625}
laminar_connectors/fix/source.rs

laminar_connectors/fix/
source.rs
