laminar_connectors/ibmmq/source/
mod.rs

1//! IBM MQ queue source over MQI with syncpoint-controlled gets.
2//!
3//! Every message a poll gets joins the connection's single unit of work. When the engine
4//! captures a checkpoint, intake pauses so that unit of work holds exactly the messages that
5//! checkpoint covers; when the checkpoint is durable, `MQCMIT` releases them and intake resumes.
6//! A crash or failed generation backs the unit of work out (explicitly, or by the queue manager
7//! when the connection breaks), so every uncommitted message is redelivered with its
8//! `BackoutCount` incremented (at-least-once; replay may duplicate).
9//!
10//! The pause lasts from barrier capture to the durable-commit notification, once per checkpoint.
11
12mod fetch;
13
14use std::sync::atomic::{AtomicU64, Ordering};
15use std::sync::Arc;
16use std::time::Duration;
17
18use arrow_schema::SchemaRef;
19use async_trait::async_trait;
20
21use super::config::IbmMqSourceConfig;
22use super::library;
23use super::metrics::SourceMetrics;
24use super::session::Session;
25use super::worker::Worker;
26use crate::checkpoint::SourceCheckpoint;
27use crate::config::ConnectorConfig;
28use crate::connector::{
29    ConnectorCancellationPolicy, ConnectorTaskOwner, ConnectorTaskTracker, SourceBatch,
30    SourceConnector, SourceConsistency, SourceContract, SourceInputMode, SourcePosition,
31    SourceStart, SourceTopology,
32};
33use crate::error::ConnectorError;
34use crate::serde::{self, RecordDeserializer};
35use fetch::{BackoutRoute, Limits, SourceSession};
36
37const GENERATION: &str = "generation";
38const GOTTEN: &str = "gotten";
39const CLOSE_TIMEOUT: Duration = Duration::from_secs(30);
40
41/// IBM MQ queue source.
42pub struct IbmMqSource {
43    schema: SchemaRef,
44    registry: Option<prometheus::Registry>,
45    running: Option<Running>,
46    failure: Option<String>,
47    /// Messages covered by the latest barrier cursor; written by `checkpoint(&self)`.
48    captured: AtomicU64,
49    task_owner: ConnectorTaskOwner,
50    task_tracker: ConnectorTaskTracker,
51}
52
53struct Running {
54    worker: Worker<SourceSession>,
55    config: Arc<IbmMqSourceConfig>,
56    generation: u64,
57    /// Messages this generation's units of work have taken from the queue.
58    gotten: u64,
59    /// Messages already committed.
60    committed: u64,
61    uow_messages: usize,
62    uow_bytes: usize,
63    uow_full: bool,
64    metrics: SourceMetrics,
65}
66
67/// What to do with a committed cursor.
68#[derive(Debug, PartialEq, Eq)]
69enum CommitPlan {
70    /// Nothing of this generation's open unit of work is covered.
71    Nothing,
72    /// The open unit of work is exactly the committed cut.
73    Commit,
74}
75
76impl Running {
77    fn cursor(&self, gotten: u64) -> SourceCheckpoint {
78        let mut cursor = SourceCheckpoint::new();
79        cursor.set_offset(GENERATION, format!("{:016x}", self.generation));
80        cursor.set_offset(GOTTEN, gotten.to_string());
81        cursor
82    }
83
84    fn plan(&self, cursor: &SourceCheckpoint) -> Result<CommitPlan, ConnectorError> {
85        plan_for(self.generation, self.gotten, self.committed, cursor)
86    }
87
88    fn limits(&self, max_records: usize) -> Option<Limits> {
89        limits_for(
90            &self.config,
91            self.uow_messages,
92            self.uow_bytes,
93            self.uow_full,
94            max_records,
95        )
96    }
97}
98
99/// A unit of work commits as a whole, so a committed cursor must name all of it.
100fn plan_for(
101    generation: u64,
102    gotten: u64,
103    committed: u64,
104    cursor: &SourceCheckpoint,
105) -> Result<CommitPlan, ConnectorError> {
106    let (Some(cursor_generation), Some(cursor_gotten)) =
107        (cursor.get_offset(GENERATION), cursor.get_offset(GOTTEN))
108    else {
109        return Ok(CommitPlan::Nothing);
110    };
111    let corrupt = || ConnectorError::InvalidState {
112        expected: "an IBM MQ source cursor".into(),
113        actual: "an unparseable generation or message count".into(),
114    };
115    let cursor_generation = u64::from_str_radix(cursor_generation, 16).map_err(|_| corrupt())?;
116    let cursor_gotten: u64 = cursor_gotten.parse().map_err(|_| corrupt())?;
117    if cursor_generation != generation || cursor_gotten <= committed {
118        return Ok(CommitPlan::Nothing);
119    }
120    if cursor_gotten != gotten {
121        return Err(ConnectorError::InvalidState {
122            expected: format!(
123                "a committed cut covering the whole unit of work ({gotten} messages)"
124            ),
125            actual: format!("a committed cut covering {cursor_gotten} messages"),
126        });
127    }
128    Ok(CommitPlan::Commit)
129}
130
131/// Poll limits within the unit of work's remaining capacity; `None` waits for a checkpoint.
132fn limits_for(
133    config: &IbmMqSourceConfig,
134    uow_messages: usize,
135    uow_bytes: usize,
136    uow_full: bool,
137    max_records: usize,
138) -> Option<Limits> {
139    let messages = config.max_uow_messages.saturating_sub(uow_messages);
140    let bytes = config.max_uow_bytes.saturating_sub(uow_bytes);
141    if uow_full || messages == 0 || bytes == 0 {
142        return None;
143    }
144    Some(Limits {
145        messages: messages.min(max_records.max(1)),
146        bytes: bytes.min(config.max_batch_bytes),
147        wait_ms: i32::try_from(config.wait_interval.as_millis()).unwrap_or(i32::MAX),
148    })
149}
150
151impl IbmMqSource {
152    /// Create an unstarted source; metrics register on `registry` at start.
153    #[must_use]
154    pub fn new(schema: SchemaRef, registry: Option<&prometheus::Registry>) -> Self {
155        let (task_owner, task_tracker) = ConnectorTaskOwner::new();
156        Self {
157            schema,
158            registry: registry.cloned(),
159            running: None,
160            failure: None,
161            captured: AtomicU64::new(0),
162            task_owner,
163            task_tracker,
164        }
165    }
166
167    /// End this generation; dropping the worker backs out the open unit of work.
168    fn fail(&mut self, error: &ConnectorError) -> ConnectorError {
169        tracing::error!(%error, "IBM MQ source generation ended");
170        self.running = None;
171        let cause = error.to_string();
172        self.failure.get_or_insert(cause.clone());
173        terminal(cause)
174    }
175
176    /// The payload codec and the payload and output schemas of the committed contract.
177    #[cfg_attr(not(feature = "fpml"), allow(clippy::unused_async))]
178    async fn codec(
179        &self,
180        config: &ConnectorConfig,
181        parsed: &IbmMqSourceConfig,
182    ) -> Result<(Box<dyn RecordDeserializer>, SchemaRef, SchemaRef), ConnectorError> {
183        #[cfg(feature = "fpml")]
184        if parsed.format == serde::Format::Fpml {
185            use crate::schema::traits::FormatDecoder;
186            let guard = self.task_owner.track().ok_or_else(retired)?;
187            let decoder = crate::fpml::message::start_decoder(
188                config,
189                config.schema_binding(),
190                "IBM MQ",
191                self.registry.as_ref(),
192                guard,
193            )
194            .await?;
195            let payload = decoder.output_schema();
196            parsed.metadata.validate_payload(&payload)?;
197            let output = parsed.metadata.output_schema(&payload);
198            let codec = crate::fpml::message::MessageDecoder(Arc::new(decoder));
199            return Ok((Box::new(codec), payload, output));
200        }
201        let (payload, output) = self.resolve_output(config, parsed)?;
202        Ok((serde::create_deserializer(parsed.format)?, payload, output))
203    }
204
205    fn resolve_output(
206        &self,
207        config: &ConnectorConfig,
208        parsed: &IbmMqSourceConfig,
209    ) -> Result<(SchemaRef, SchemaRef), ConnectorError> {
210        let declared = config
211            .arrow_schema()
212            .or_else(|| (!self.schema.fields().is_empty()).then(|| Arc::clone(&self.schema)));
213        let payload = declared
214            .map(|output| parsed.metadata.payload_schema(&output))
215            .transpose()?;
216        let binding = serde::schema_contract::reader_binding(config, parsed.format, payload)?;
217        let payload = Arc::new(binding.logical);
218        parsed.metadata.validate_payload(&payload)?;
219        let output = parsed.metadata.output_schema(&payload);
220        if config
221            .schema_binding()
222            .is_some_and(|expected| expected.logical != *output)
223        {
224            return Err(ConnectorError::SchemaMismatch(
225                "IBM MQ reader differs from its committed contract".into(),
226            ));
227        }
228        Ok((payload, output))
229    }
230}
231
232/// Open the input queue and validate the queue's backout settings before any get.
233fn open_session(
234    config: &Arc<IbmMqSourceConfig>,
235    library: library::Library,
236    deserializer: Box<dyn RecordDeserializer>,
237    payload: SchemaRef,
238    output: SchemaRef,
239) -> Result<SourceSession, ConnectorError> {
240    use mqi::constants;
241    let session = Session::open(
242        &config.connection,
243        library,
244        &config.queue,
245        None,
246        constants::MQOO_INPUT_AS_Q_DEF | constants::MQOO_INQUIRE,
247    )?;
248    let (threshold, queue) = session.backout_settings()?;
249    let route = match (threshold, queue.is_empty()) {
250        (0, _) => None,
251        (_, true) => {
252            return Err(ConnectorError::ConfigurationError(format!(
253                "queue '{}' sets BOTHRESH({threshold}) without BOQNAME; set BOQNAME to a local \
254                 queue or BOTHRESH(0). The connector never falls back to the dead-letter queue",
255                config.queue
256            )))
257        }
258        (_, false) => {
259            let resolved = session.resolve(&queue)?;
260            if !resolved.is_empty() && resolved != session.queue_manager {
261                return Err(ConnectorError::ConfigurationError(format!(
262                    "BOQNAME '{queue}' resolves to queue manager '{resolved}', not '{}'; a poison \
263                     message moves atomically only to a local queue",
264                    session.queue_manager
265                )));
266            }
267            Some(BackoutRoute { threshold, queue })
268        }
269    };
270    if route.is_none() && config.on_decode_error == super::config::DecodeErrorPolicy::BackoutQueue {
271        return Err(ConnectorError::ConfigurationError(format!(
272            "on.decode.error = 'backout-queue' needs queue '{}' to set BOTHRESH and a local BOQNAME",
273            config.queue
274        )));
275    }
276    Ok(SourceSession::new(
277        session,
278        route,
279        Arc::clone(config),
280        deserializer,
281        payload,
282        output,
283    ))
284}
285
286fn retired() -> ConnectorError {
287    ConnectorError::Internal("IBM MQ source task generation is retired".into())
288}
289
290fn terminal(cause: String) -> ConnectorError {
291    ConnectorError::InvalidState {
292        expected: "a live IBM MQ connection".into(),
293        actual: cause,
294    }
295}
296
297#[async_trait]
298impl SourceConnector for IbmMqSource {
299    /// Dropping a poll or commit leaves the source valid: a poll whose result is lost after it
300    /// got messages taints the unit of work, which then never commits and is backed out at close.
301    /// The ledger only advances after an awaited result arrives.
302    fn cancellation_policy(&self) -> ConnectorCancellationPolicy {
303        ConnectorCancellationPolicy::CancelSafe
304    }
305
306    fn terminal_task_tracker(&self) -> Option<ConnectorTaskTracker> {
307        Some(self.task_tracker.clone())
308    }
309
310    async fn resolve_schema(
311        &mut self,
312        config: &ConnectorConfig,
313        explicit: Option<SchemaRef>,
314    ) -> Result<crate::schema::resolution::SchemaBinding, ConnectorError> {
315        use crate::schema::resolution::{logical_binding, SchemaDirection};
316        let parsed = IbmMqSourceConfig::from_config(config)?;
317        #[cfg(feature = "fpml")]
318        if parsed.format == serde::Format::Fpml {
319            return crate::fpml::message::source_binding(config, explicit, "IBM MQ", |payload| {
320                parsed.metadata.validate_payload(payload)?;
321                Ok(parsed.metadata.output_schema(payload))
322            })
323            .await;
324        }
325        let reader = serde::schema_contract::reader_binding(config, parsed.format, explicit)?;
326        let payload = Arc::new(reader.logical);
327        parsed.metadata.validate_payload(&payload)?;
328        logical_binding(
329            config,
330            SchemaDirection::Source,
331            reader.origin,
332            &parsed.metadata.output_schema(&payload),
333        )
334    }
335
336    fn contract(&self, config: &ConnectorConfig) -> Result<SourceContract, ConnectorError> {
337        IbmMqSourceConfig::from_config(config)?;
338        // MQCMIT runs only for a durable checkpoint. Exactly-once is not certified: the queue
339        // commit and the sink commit are separate.
340        Ok(SourceContract::new(
341            SourceConsistency::CommitCoupled,
342            SourceTopology::Singleton,
343            SourceInputMode::AppendOnly,
344        ))
345    }
346
347    fn recovery_identity_options(
348        &self,
349        config: &ConnectorConfig,
350    ) -> Result<Option<std::collections::BTreeMap<String, String>>, ConnectorError> {
351        Ok(Some(IbmMqSourceConfig::from_config(config)?.identity()))
352    }
353
354    async fn start(&mut self, request: SourceStart) -> Result<(), ConnectorError> {
355        let (config, position, _) = request.into_parts();
356        if matches!(position, SourcePosition::Initialized { .. }) {
357            return Err(ConnectorError::ConfigurationError(
358                "IBM MQ has no sealed topology startup contract".into(),
359            ));
360        }
361        // A resume needs no positioning: the queue manager backed out every message the
362        // committed cut did not cover when the previous connection ended.
363        let parsed = Arc::new(IbmMqSourceConfig::from_config(&config)?);
364        let (deserializer, payload, output) = self.codec(&config, &parsed).await?;
365        let library = library::load(parsed.connection.client_library.as_deref())?;
366        let name = config
367            .get("laminar.source.name")
368            .unwrap_or("ibmmq")
369            .to_string();
370        let metrics = SourceMetrics::new(&name, self.registry.as_ref())?;
371        let guard = self.task_owner.track().ok_or_else(retired)?;
372        let session_config = Arc::clone(&parsed);
373        let session_output = Arc::clone(&output);
374        let worker = Worker::start(
375            &format!("ibmmq-src-{name}"),
376            guard,
377            parsed.connection.operation_timeout,
378            move || {
379                open_session(
380                    &session_config,
381                    library,
382                    deserializer,
383                    payload,
384                    session_output,
385                )
386            },
387        )
388        .await?;
389        tracing::info!(
390            queue = %parsed.queue,
391            connection = ?parsed.connection,
392            "IBM MQ source consuming under syncpoint"
393        );
394        self.schema = output;
395        self.failure = None;
396        self.captured.store(0, Ordering::Relaxed);
397        self.running = Some(Running {
398            worker,
399            config: parsed,
400            generation: rand::random(),
401            gotten: 0,
402            committed: 0,
403            uow_messages: 0,
404            uow_bytes: 0,
405            uow_full: false,
406            metrics,
407        });
408        Ok(())
409    }
410
411    async fn poll_batch(
412        &mut self,
413        max_records: usize,
414    ) -> Result<Option<SourceBatch>, ConnectorError> {
415        if let Some(cause) = &self.failure {
416            return Err(terminal(cause.clone()));
417        }
418        let Some(running) = self.running.as_mut() else {
419            return Ok(None);
420        };
421        // A captured cut not yet committed seals the unit of work: getting more now would
422        // commit messages that checkpoint does not contain.
423        if self.captured.load(Ordering::Relaxed) > running.committed {
424            return Ok(None);
425        }
426        let Some(limits) = running.limits(max_records) else {
427            return Ok(None);
428        };
429        let timeout = running.config.connection.operation_timeout + running.config.wait_interval;
430        let polled = match running
431            .worker
432            .call_or(
433                "MQGET",
434                timeout,
435                move |session| session.poll(&limits),
436                |session, polled| session.lost(&polled),
437            )
438            .await
439        {
440            Ok(polled) => polled,
441            Err(error) => return Err(self.fail(&error)),
442        };
443        running.gotten += polled.gotten as u64;
444        running.uow_messages += polled.gotten;
445        running.uow_bytes += polled.bytes;
446        running.uow_full |= polled.uow_full;
447        let metrics = &running.metrics;
448        metrics.messages.inc_by(polled.gotten as u64);
449        metrics.bytes.inc_by(polled.bytes as u64);
450        metrics.moved.inc_by(polled.moved as u64);
451        metrics.redelivered.inc_by(polled.redelivered as u64);
452        metrics
453            .uncommitted
454            .set(i64::try_from(running.uow_messages).unwrap_or(i64::MAX));
455        let records = polled.records.filter(|records| records.num_rows() > 0);
456        if let Some(records) = &records {
457            metrics.rows.inc_by(records.num_rows() as u64);
458        }
459        // A bound cursor keeps the runtime from capturing through `checkpoint`, which seals
460        // the unit of work; only a barrier capture should.
461        let cursor = running.cursor(running.gotten);
462        Ok(records.map(|records| SourceBatch::new(records).with_checkpoint(cursor)))
463    }
464
465    fn schema(&self) -> SchemaRef {
466        Arc::clone(&self.schema)
467    }
468
469    fn checkpoint(&self) -> SourceCheckpoint {
470        let Some(running) = self.running.as_ref() else {
471            return SourceCheckpoint::new();
472        };
473        self.captured.store(running.gotten, Ordering::Relaxed);
474        running.cursor(running.gotten)
475    }
476
477    async fn notify_epoch_committed(
478        &mut self,
479        epoch: u64,
480        checkpoint: &SourceCheckpoint,
481    ) -> Result<(), ConnectorError> {
482        let Some(running) = self.running.as_mut() else {
483            return Ok(());
484        };
485        let plan = match running.plan(checkpoint) {
486            Ok(plan) => plan,
487            Err(error) => return Err(self.fail(&error)),
488        };
489        if plan == CommitPlan::Nothing {
490            return Ok(());
491        }
492        let timeout = running.config.connection.operation_timeout;
493        if let Err(error) = running
494            .worker
495            .call("MQCMIT", timeout, |session| session.commit())
496            .await
497        {
498            return Err(self.fail(&ConnectorError::outcome_unknown(
499                format!(
500                    "MQCMIT for committed epoch {epoch}: {error}; its messages may be redelivered"
501                ),
502                false,
503            )));
504        }
505        running
506            .metrics
507            .committed
508            .inc_by(running.gotten - running.committed);
509        running.committed = running.gotten;
510        running.uow_messages = 0;
511        running.uow_bytes = 0;
512        running.uow_full = false;
513        running.metrics.uncommitted.set(0);
514        Ok(())
515    }
516
517    async fn close(&mut self) -> Result<(), ConnectorError> {
518        let Some(running) = self.running.take() else {
519            return Ok(());
520        };
521        running.worker.stop(CLOSE_TIMEOUT).await
522    }
523}
524
525#[cfg(test)]
526mod tests;
laminar_connectors/ibmmq/source/mod.rs

laminar_connectors/ibmmq/source/
mod.rs
