laminar_db/pipeline/streaming_coordinator/
construction.rs

1//! Mechanically extracted coordinator responsibility.
2
3use super::source_actor::SourceActorSpawner;
4#[cfg(all(test, feature = "cluster"))]
5use super::OwnedConnectorTaskFences;
6use super::{
7    admit_append_only_source, run_source_operation, schema_has_reserved_mutation_columns,
8    start_source_once, Arc, AtomicU64, CheckpointAttempt, CheckpointBarrier,
9    ConnectorCancellationPolicy, ControlMsgRx, DbError, DeliveryGuarantee, Duration, FxHashMap,
10    FxHashSet, Instant, OwnedSourceTasks, PendingBarrier, PipelineConfig, PreparedSourceGeneration,
11    SourceAdmission, SourceCheckpoint, SourceConsistency, SourceInputMode, SourceOperationOutcome,
12    SourcePosition, SourceRegistration, SourceRowPositionCapability, SourceStart,
13    SourceStartFailure, SourceStartOutcome, SourceTaskLease, StreamingCoordinator,
14    StreamingCoordinatorRuntime, TrackedSourceRegistration,
15};
16#[cfg(feature = "cluster")]
17use super::{
18    source_process_authority_is_live, ClusterController, CycleError, SourceProcessAuthority,
19};
20
21struct PreparedSourceSet {
22    prepared_sources: Vec<PreparedSourceGeneration>,
23    committed_offsets: Vec<Option<SourceCheckpoint>>,
24}
25impl StreamingCoordinator {
26    pub(crate) fn observe_flow_metrics(&self, metrics: &crate::EngineMetrics) {
27        metrics.flow.coordinator.observe(
28            self.rx.observation(self.config.source_queue_max_bytes),
29            Arc::clone(&self.intake_stats),
30        );
31    }
32
33    pub(super) fn admit_public_source_shapes(
34        sources: &[TrackedSourceRegistration],
35    ) -> Result<(), DbError> {
36        for source in sources {
37            if source.expected_schema.fields().is_empty() {
38                return Err(DbError::Config(format!(
39                    "source '{}' must expose a non-empty schema before public coordinator startup; late-bound schemas require database-owned catalog admission",
40                    source.name
41                )));
42            }
43            if let Some(mode) = source.admitted_non_append_mode {
44                if source.contract().input_mode != mode
45                    || source.contract().row_positions
46                        != SourceRowPositionCapability::OrderedDeterministic
47                {
48                    return Err(DbError::Config(format!(
49                        "source '{}' lost its admitted stateful mutation contract",
50                        source.name
51                    )));
52                }
53                match mode {
54                    SourceInputMode::AppendOnly => {
55                        return Err(DbError::Config(format!(
56                            "source '{}' has an invalid append-only mutation admission marker",
57                            source.name
58                        )));
59                    }
60                    SourceInputMode::KeyedUpsert => {
61                        if source.has_reserved_mutation_columns() {
62                            return Err(DbError::Config(format!(
63                                "source '{}' keyed-upsert schema declares reserved mutation metadata",
64                                source.name
65                            )));
66                        }
67                    }
68                    SourceInputMode::FullChangelog => {
69                        let weight = laminar_core::changelog::WEIGHT_COLUMN;
70                        let fields = source.expected_schema.fields();
71                        let valid_weight = fields.last().is_some_and(|field| {
72                            field.name() == weight
73                                && field.data_type() == &arrow_schema::DataType::Int64
74                                && !field.is_nullable()
75                        });
76                        let reserved_count = fields
77                            .iter()
78                            .filter(|field| {
79                                ["_op", "__op", weight]
80                                    .iter()
81                                    .any(|name| field.name().eq_ignore_ascii_case(name))
82                            })
83                            .count();
84                        if !valid_weight || reserved_count != 1 {
85                            return Err(DbError::Config(format!(
86                                "source '{}' full-changelog schema requires exact trailing non-null Int64 '{weight}'",
87                                source.name
88                            )));
89                        }
90                    }
91                }
92            } else {
93                admit_append_only_source(
94                    source.contract(),
95                    source.has_reserved_mutation_columns(),
96                )
97                .map_err(|reason| {
98                    DbError::Config(format!(
99                        "source '{}' is not admissible through the public coordinator: {reason} (contract: {:?})",
100                        source.name,
101                        source.contract()
102                    ))
103                })?;
104            }
105        }
106        Ok(())
107    }
108
109    #[cfg(feature = "cluster")]
110    #[inline]
111    pub(super) fn require_process_authority(&self, boundary: &str) -> Result<(), CycleError> {
112        if self
113            .process_authority
114            .as_deref()
115            .is_none_or(SourceProcessAuthority::is_live)
116        {
117            Ok(())
118        } else {
119            Err(CycleError::Recovery(format!(
120                "cluster process lease expired before {boundary}"
121            )))
122        }
123    }
124
125    pub(super) async fn close_startup_source(
126        source: &mut SourceRegistration,
127        cleanup_deadline: tokio::time::Instant,
128        #[cfg(feature = "cluster")] process_authority: Option<&SourceProcessAuthority>,
129    ) {
130        match run_source_operation(
131            cleanup_deadline,
132            #[cfg(feature = "cluster")]
133            process_authority,
134            || source.connector.close(),
135        )
136        .await
137        {
138            SourceOperationOutcome::Completed(Ok(())) => {}
139            SourceOperationOutcome::Completed(Err(error)) => {
140                tracing::warn!(
141                    source = %source.name,
142                    %error,
143                    "source close failed while rolling back pipeline startup"
144                );
145            }
146            SourceOperationOutcome::Deadline => {
147                tracing::warn!(
148                    source = %source.name,
149                    "source close exceeded its pipeline-startup cleanup deadline"
150                );
151            }
152            #[cfg(feature = "cluster")]
153            SourceOperationOutcome::ProcessAuthorityLost => {}
154        }
155    }
156
157    pub(super) async fn close_prepared_sources(
158        sources: &mut Vec<PreparedSourceGeneration>,
159        cleanup_timeout: Duration,
160        #[cfg(feature = "cluster")] process_authority: Option<&SourceProcessAuthority>,
161    ) {
162        let cleanup_deadline = tokio::time::Instant::now() + cleanup_timeout;
163        futures::future::join_all(sources.iter_mut().map(|source| {
164            Self::close_startup_source(
165                &mut source.registration,
166                cleanup_deadline,
167                #[cfg(feature = "cluster")]
168                process_authority,
169            )
170        }))
171        .await;
172        sources.clear();
173    }
174
175    /// Reap a source task while preserving its generation fence until actual termination.
176    pub(super) fn reap_source_task(task: SourceTaskLease) {
177        if !task.is_finished() {
178            tracing::warn!(
179                source = %task.name(),
180                "source task did not exit within shutdown budget; retiring its connector generation"
181            );
182            task.abort();
183            // The DB-owned lease remains registered until the actor wrapper and connector tracker
184            // prove actual termination. Coordinator shutdown stays bounded without admitting an
185            // overlapping replacement generation.
186            drop(task);
187            return;
188        }
189        task.log_terminal_outcome();
190        drop(task);
191    }
192
193    /// Notify every source of a committed epoch so it can release retained upstream data.
194    pub(super) fn broadcast_epoch_committed(
195        &self,
196        epoch: u64,
197        per_source: &FxHashMap<String, SourceCheckpoint>,
198    ) {
199        for handle in &self.source_handles {
200            let cp = per_source
201                .get(handle.task.name())
202                .cloned()
203                .unwrap_or_else(SourceCheckpoint::new);
204            let _ = handle.epoch_committed_tx.send(Some((epoch, cp)));
205        }
206    }
207
208    pub(super) fn release_source_barrier_attempt(&self, attempt: CheckpointAttempt) {
209        for handle in &self.source_handles {
210            handle.barrier_control().release_exact(attempt);
211        }
212    }
213
214    pub(super) fn release_source_barrier_for(&self, source_idx: usize, attempt: CheckpointAttempt) {
215        if let Some(handle) = self.source_handles.get(source_idx) {
216            handle.barrier_control().release_exact(attempt);
217        }
218    }
219
220    pub(super) fn stop_source_barrier_holds(&self) {
221        for handle in &self.source_handles {
222            handle.barrier_control().stop_hold();
223        }
224    }
225
226    pub(super) fn cancel_local_source_barriers(&self, barrier: CheckpointBarrier) {
227        for handle in &self.source_handles {
228            handle.barrier_control().cancel_exact(barrier);
229        }
230    }
231
232    /// Build the coordinator, atomically start each source connector, and spawn source tasks.
233    ///
234    /// # Errors
235    ///
236    /// Returns an error if delivery guarantee constraints are violated or a source fails to start
237    /// at its requested initial/recovered position.
238    pub async fn new(
239        runtime: &StreamingCoordinatorRuntime,
240        sources: Vec<SourceRegistration>,
241        config: PipelineConfig,
242        shutdown: Arc<tokio::sync::Notify>,
243        control_rx: ControlMsgRx,
244        source_gate: Arc<std::sync::atomic::AtomicBool>,
245    ) -> Result<Self, DbError> {
246        let _construction = runtime.construction.lock().await;
247        runtime.prune_and_require_idle()?;
248        let generation = runtime.claim_generation()?;
249        let sources = sources
250            .into_iter()
251            .map(|source| {
252                TrackedSourceRegistration::capture(source, &runtime.owned_connector_task_fences)
253            })
254            .collect::<Result<Vec<_>, _>>()?;
255        Self::admit_public_source_shapes(&sources)?;
256        if let Some(source) = sources.iter().find(|source| source.assignment_scoped) {
257            return Err(DbError::Config(format!(
258                "assignment-scoped source '{}' requires the database-owned cluster runtime",
259                source.name
260            )));
261        }
262        let mut coordinator = Self::new_with_tracked_source_registry(
263            sources,
264            config,
265            shutdown,
266            control_rx,
267            source_gate,
268            #[cfg(feature = "cluster")]
269            None,
270            Arc::clone(&runtime.owned_source_tasks),
271            crate::db::RuntimeMode::Local,
272        )
273        .await?;
274        coordinator.public_generation = Some(generation);
275        Ok(coordinator)
276    }
277
278    #[cfg(all(test, feature = "cluster"))]
279    pub(crate) async fn new_with_source_registry(
280        sources: Vec<SourceRegistration>,
281        config: PipelineConfig,
282        shutdown: Arc<tokio::sync::Notify>,
283        control_rx: ControlMsgRx,
284        source_gate: Arc<std::sync::atomic::AtomicBool>,
285        #[cfg(feature = "cluster")] source_process_authority: Option<Arc<ClusterController>>,
286        owned_source_tasks: OwnedSourceTasks,
287        owned_connector_task_fences: OwnedConnectorTaskFences,
288        runtime_mode: crate::db::RuntimeMode,
289    ) -> Result<Self, DbError> {
290        let sources = sources
291            .into_iter()
292            .map(|source| TrackedSourceRegistration::capture(source, &owned_connector_task_fences))
293            .collect::<Result<Vec<_>, _>>()?;
294        Self::admit_public_source_shapes(&sources)?;
295        Self::new_with_tracked_source_registry(
296            sources,
297            config,
298            shutdown,
299            control_rx,
300            source_gate,
301            #[cfg(feature = "cluster")]
302            source_process_authority,
303            owned_source_tasks,
304            runtime_mode,
305        )
306        .await
307    }
308
309    fn validate_source_runtime_config(
310        sources: &[TrackedSourceRegistration],
311        config: &PipelineConfig,
312        #[cfg(feature = "cluster")] source_process_authority: Option<&ClusterController>,
313        #[cfg(feature = "cluster")] runtime_mode: crate::db::RuntimeMode,
314    ) -> Result<(), DbError> {
315        if config
316            .checkpoint_schedule
317            .periodic_interval()
318            .is_some_and(|interval| interval.is_zero())
319        {
320            return Err(DbError::Config(
321                "checkpoint interval must be greater than zero; use manual checkpointing instead"
322                    .into(),
323            ));
324        }
325        if config.delivery_guarantee == DeliveryGuarantee::BestEffort {
326            for src in sources {
327                if src.contract().consistency == SourceConsistency::CommitCoupled {
328                    return Err(DbError::Config(format!(
329                        "source '{}' is commit-coupled; commit-coupled sources support only at-least-once and certified exactly-once delivery",
330                        src.name
331                    )));
332                }
333            }
334        }
335        if config.delivery_guarantee == DeliveryGuarantee::ExactlyOnce {
336            for src in sources {
337                if !src.contract().is_exact_delivery_certified() {
338                    return Err(DbError::Config(format!(
339                        "[{}] exactly-once source '{}' is not production-certified",
340                        laminar_core::error_codes::EXACTLY_ONCE_SOURCE_UNCERTIFIED,
341                        src.name
342                    )));
343                }
344            }
345        }
346        if matches!(
347            config.delivery_guarantee,
348            DeliveryGuarantee::AtLeastOnce | DeliveryGuarantee::ExactlyOnce
349        ) {
350            for src in sources {
351                if !src.contract().supports_replay() {
352                    return Err(DbError::Config(format!(
353                        "[LDB-5031] {} requires source '{}' to support replay",
354                        config.delivery_guarantee, src.name
355                    )));
356                }
357            }
358            if !config.checkpoint_schedule.is_enabled() {
359                return Err(DbError::Config(format!(
360                    "[LDB-5032] {} requires checkpointing to be enabled",
361                    config.delivery_guarantee
362                )));
363            }
364        }
365
366        // A source that releases externally retained data only on durable commit needs
367        // checkpointing; otherwise that data can grow without bound. Reject the combination up
368        // front.
369        if !config.checkpoint_schedule.is_enabled() {
370            for src in sources {
371                if src.contract().requires_checkpointing() {
372                    return Err(DbError::Config(format!(
373                        "[LDB-5034] source '{}' requires checkpointing to be enabled: externally \
374                         retained data is only released at a durable checkpoint",
375                        src.name
376                    )));
377                }
378            }
379        }
380
381        if config.channel_capacity == 0 {
382            return Err(DbError::Config(
383                "[LDB-0010] channel_capacity must be > 0".into(),
384            ));
385        }
386        crate::config::validate_source_queue_max_bytes(config.source_queue_max_bytes)
387            .map_err(|error| DbError::Config(error.into()))?;
388
389        #[cfg(feature = "cluster")]
390        if runtime_mode.is_cluster() {
391            let controller = source_process_authority.as_ref().ok_or_else(|| {
392                DbError::Config(
393                    "cluster source runtime requires a cluster controller with process lease authority"
394                        .into(),
395                )
396            })?;
397            if controller.process_lease_deadline().is_none() {
398                return Err(DbError::Config(
399                    "cluster source runtime requires one shared process lease deadline before construction"
400                        .into(),
401                ));
402            }
403        } else if source_process_authority.is_some() {
404            return Err(DbError::Config(
405                "local source runtime cannot install cluster process lease authority".into(),
406            ));
407        }
408
409        Ok(())
410    }
411
412    fn prepare_source_starts(
413        sources: Vec<TrackedSourceRegistration>,
414        delivery_guarantee: DeliveryGuarantee,
415    ) -> Result<Vec<(TrackedSourceRegistration, SourceStart)>, DbError> {
416        let mut source_starts = Vec::with_capacity(sources.len());
417        for src in sources {
418            let start = SourceStart::new(
419                src.config.clone(),
420                src.position.clone(),
421                delivery_guarantee,
422            )
423            .map_err(|error| match &src.position {
424                SourcePosition::Initialized { .. } => DbError::Checkpoint(format!(
425                    "source '{}' has an invalid sealed initialization request: {error}",
426                    src.name
427                )),
428                SourcePosition::Initial => DbError::Config(format!(
429                    "source '{}' has an invalid initial startup request: {error}",
430                    src.name
431                )),
432                SourcePosition::Resume { attempt, .. } => DbError::Checkpoint(format!(
433                    "[LDB-6003] source '{}' has an invalid resume request for checkpoint epoch={} id={}: {error}",
434                    src.name, attempt.epoch, attempt.checkpoint_id
435                )),
436            })?;
437            if matches!(&src.position, SourcePosition::Initialized { .. })
438                && !src.connector.supports_initialized_start()
439            {
440                return Err(DbError::Config(format!(
441                    "source '{}' connector has no certified sealed initialization startup contract",
442                    src.name
443                )));
444            }
445            source_starts.push((src, start));
446        }
447        Ok(source_starts)
448    }
449
450    async fn prepare_source_generations(
451        source_starts: Vec<(TrackedSourceRegistration, SourceStart)>,
452        source_start_timeout: Duration,
453        #[cfg(feature = "cluster")] source_process_authority: Option<&SourceProcessAuthority>,
454    ) -> Result<PreparedSourceSet, DbError> {
455        let source_count = source_starts.len();
456        let mut prepared_sources = Vec::with_capacity(source_count);
457        let mut committed_offsets = Vec::with_capacity(source_count);
458        let source_start_deadline = tokio::time::Instant::now() + source_start_timeout;
459
460        // Do not spawn a polling task until every source has atomically installed its startup
461        // position. Otherwise a later startup failure detaches the earlier tasks and they keep
462        // polling without an owner capable of shutting them down.
463        for (mut src, start) in source_starts {
464            let src_name = src.name.clone();
465            let start_position = src.position.clone();
466            if tokio::time::Instant::now() >= source_start_deadline {
467                #[cfg(feature = "cluster")]
468                if !source_process_authority_is_live(source_process_authority) {
469                    return Err(DbError::Connector(format!(
470                        "source '{src_name}' start was not attempted: cluster process lease expired"
471                    )));
472                }
473                // `timeout_at` polls its inner future once even when already expired. Starting a
474                // source can acquire a lease/slot, so never construct or poll it after the shared
475                // stage budget has been consumed by earlier sources.
476                Self::close_prepared_sources(
477                    &mut prepared_sources,
478                    PipelineConfig::CONNECTOR_STARTUP_CLEANUP_TIMEOUT,
479                    #[cfg(feature = "cluster")]
480                    source_process_authority,
481                )
482                .await;
483                let error = format!(
484                    "shared {source_start_timeout:?} source-start stage deadline exhausted before start began"
485                );
486                return Err(source_stage_deadline_error(
487                    &src_name,
488                    &start_position,
489                    &error,
490                ));
491            }
492            // Only a durable resume seeds committed progress. Sealed initialization is a start
493            // boundary, not processed input and never an acknowledgement of the skipped prefix.
494            // Capture it before moving the complete request into `start`; no connector lifecycle
495            // operation is allowed between configuration and cursor installation.
496            let committed_offset = match &src.position {
497                SourcePosition::Initial | SourcePosition::Initialized { .. } => None,
498                SourcePosition::Resume { checkpoint, .. } => Some(checkpoint.clone()),
499            };
500            let cancellation_policy = src.connector.cancellation_policy();
501            let source_start_authorized = {
502                #[cfg(feature = "cluster")]
503                {
504                    source_process_authority_is_live(source_process_authority)
505                }
506                #[cfg(not(feature = "cluster"))]
507                {
508                    true
509                }
510            };
511            let mut start_error = if source_start_authorized {
512                match start_source_once(
513                    src.connector.as_mut(),
514                    start,
515                    source_start_deadline,
516                    #[cfg(feature = "cluster")]
517                    source_process_authority,
518                )
519                .await
520                {
521                    SourceStartOutcome::Completed(Ok(())) => {
522                        #[cfg(feature = "cluster")]
523                        if source_process_authority_is_live(source_process_authority) {
524                            None
525                        } else {
526                            Some(SourceStartFailure::ProcessAuthorityLost(
527                                "process lease expired as source start completed".to_owned(),
528                            ))
529                        }
530                        #[cfg(not(feature = "cluster"))]
531                        None
532                    }
533                    SourceStartOutcome::Completed(Err(error)) => {
534                        Some(if error.is_outcome_unknown() {
535                            SourceStartFailure::Retired(error.to_string())
536                        } else {
537                            SourceStartFailure::Connector(error.to_string())
538                        })
539                    }
540                    SourceStartOutcome::TimedOut => Some(
541                        if cancellation_policy == ConnectorCancellationPolicy::RetireConnector {
542                            SourceStartFailure::Retired(format!(
543                                "exceeded the shared {source_start_timeout:?} source-start stage deadline"
544                            ))
545                        } else {
546                            SourceStartFailure::Connector(format!(
547                                "exceeded the shared {source_start_timeout:?} source-start stage deadline"
548                            ))
549                        },
550                    ),
551                    #[cfg(feature = "cluster")]
552                    SourceStartOutcome::ProcessAuthorityLost => {
553                        Some(SourceStartFailure::ProcessAuthorityLost(
554                            "process lease expired while source start was in flight".to_owned(),
555                        ))
556                    }
557                }
558            } else {
559                #[cfg(feature = "cluster")]
560                {
561                    Some(SourceStartFailure::ProcessAuthorityLost(
562                        "process lease expired before source start began".to_owned(),
563                    ))
564                }
565                #[cfg(not(feature = "cluster"))]
566                {
567                    unreachable!("local source startup is always authorized")
568                }
569            };
570            if start_error.is_none() {
571                let started_schema = src.connector.schema();
572                if src.schema_admitted {
573                    if started_schema.as_ref() != src.expected_schema.as_ref() {
574                        start_error = Some(SourceStartFailure::Connector(format!(
575                            "schema after start does not match the admitted schema for source '{src_name}'"
576                        )));
577                    }
578                } else {
579                    src.expected_schema = started_schema;
580                    if let Err(reason) = admit_append_only_source(
581                        src.contract,
582                        schema_has_reserved_mutation_columns(src.expected_schema.as_ref()),
583                    ) {
584                        start_error = Some(SourceStartFailure::Connector(format!(
585                            "source '{src_name}' schema after start is not admissible: {reason}"
586                        )));
587                    }
588                }
589                if start_error.is_none() {
590                    match TrackedSourceRegistration::metadata_schemas(
591                        &src_name,
592                        src.contract,
593                        &src.expected_schema,
594                    ) {
595                        Ok((positioned, mutations)) => {
596                            src.positioned_schema = positioned;
597                            src.mutation_schema = mutations;
598                        }
599                        Err(error) => {
600                            start_error = Some(SourceStartFailure::Connector(error.to_string()));
601                        }
602                    }
603                }
604            }
605            if let Some(failure) = start_error {
606                let error = match failure {
607                    SourceStartFailure::Connector(error) => {
608                        // A completed failure may have acquired resources, but the connector is
609                        // still valid for its bounded terminal cleanup.
610                        prepared_sources.push(PreparedSourceGeneration { registration: src });
611                        Self::close_prepared_sources(
612                            &mut prepared_sources,
613                            PipelineConfig::CONNECTOR_STARTUP_CLEANUP_TIMEOUT,
614                            #[cfg(feature = "cluster")]
615                            source_process_authority,
616                        )
617                        .await;
618                        error
619                    }
620                    SourceStartFailure::Retired(error) => {
621                        // The cancelled current generation is terminal. Drop it without invoking
622                        // close and clean up only sources whose starts completed.
623                        Self::close_prepared_sources(
624                            &mut prepared_sources,
625                            PipelineConfig::CONNECTOR_STARTUP_CLEANUP_TIMEOUT,
626                            #[cfg(feature = "cluster")]
627                            source_process_authority,
628                        )
629                        .await;
630                        error
631                    }
632                    #[cfg(feature = "cluster")]
633                    SourceStartFailure::ProcessAuthorityLost(error) => {
634                        // Generic close may publish externally. Drop every generation without
635                        // another connector call after the process authority fence.
636                        error
637                    }
638                };
639                return match start_position {
640                    SourcePosition::Initialized { .. } => Err(DbError::Checkpoint(format!(
641                        "source '{src_name}' sealed initialization failed: {error}"
642                    ))),
643                    SourcePosition::Initial => Err(DbError::Config(format!(
644                        "source '{src_name}' start failed at initial position: {error}"
645                    ))),
646                    SourcePosition::Resume { attempt, .. } => Err(DbError::Checkpoint(format!(
647                        "[LDB-6003] source '{src_name}' start failed while resuming exact \
648                             checkpoint epoch={} id={}: {error}",
649                        attempt.epoch, attempt.checkpoint_id
650                    ))),
651                };
652            }
653
654            committed_offsets.push(committed_offset);
655            prepared_sources.push(PreparedSourceGeneration { registration: src });
656        }
657
658        Ok(PreparedSourceSet {
659            prepared_sources,
660            committed_offsets,
661        })
662    }
663    pub(crate) async fn new_with_tracked_source_registry(
664        sources: Vec<TrackedSourceRegistration>,
665        config: PipelineConfig,
666        shutdown: Arc<tokio::sync::Notify>,
667        control_rx: ControlMsgRx,
668        source_gate: Arc<std::sync::atomic::AtomicBool>,
669        #[cfg(feature = "cluster")] source_process_authority: Option<Arc<ClusterController>>,
670        owned_source_tasks: OwnedSourceTasks,
671        #[cfg(feature = "cluster")] runtime_mode: crate::db::RuntimeMode,
672        #[cfg(not(feature = "cluster"))] _runtime_mode: crate::db::RuntimeMode,
673    ) -> Result<Self, DbError> {
674        Self::validate_source_runtime_config(
675            &sources,
676            &config,
677            #[cfg(feature = "cluster")]
678            source_process_authority.as_deref(),
679            #[cfg(feature = "cluster")]
680            runtime_mode,
681        )?;
682
683        #[cfg(feature = "cluster")]
684        let source_process_authority = source_process_authority.map(SourceProcessAuthority::new);
685        let source_starts = Self::prepare_source_starts(sources, config.delivery_guarantee)?;
686        let PreparedSourceSet {
687            prepared_sources,
688            committed_offsets,
689        } = Self::prepare_source_generations(
690            source_starts,
691            config.checkpoint_timeout,
692            #[cfg(feature = "cluster")]
693            source_process_authority.as_deref(),
694        )
695        .await?;
696        let source_count = prepared_sources.len();
697        let (tx, rx) =
698            super::source_channel::channel(config.channel_capacity, config.source_queue_max_bytes);
699        let (source_fault_tx, source_fault_rx) = tokio::sync::mpsc::unbounded_channel();
700        let mut source_handles = Vec::with_capacity(source_count);
701        let mut source_names = Vec::with_capacity(source_count);
702        let mut source_input_modes = Vec::with_capacity(source_count);
703        let source_runtime = tokio::runtime::Handle::current();
704
705        let spawner = SourceActorSpawner {
706            tx: &tx,
707            source_fault_tx: &source_fault_tx,
708            source_gate: &source_gate,
709            config: &config,
710            runtime: &source_runtime,
711            owned_source_tasks: &owned_source_tasks,
712            #[cfg(feature = "cluster")]
713            source_process_authority: source_process_authority.as_ref(),
714            #[cfg(feature = "cluster")]
715            runtime_mode,
716        };
717        for (idx, prepared) in prepared_sources.into_iter().enumerate() {
718            let source_actor = spawner.spawn(idx, prepared);
719            source_handles.push(source_actor.handle);
720            source_names.push(source_actor.name);
721            source_input_modes.push(source_actor.input_mode);
722        }
723        Ok(Self {
724            config,
725            rx,
726            source_fault_rx,
727            source_handles,
728            source_names,
729            source_input_modes,
730            shutdown,
731            terminal_shutdown: tokio_util::sync::CancellationToken::new(),
732            pending_barrier: PendingBarrier::new(),
733            last_checkpoint: Instant::now(),
734            checkpoint_retry_not_before: None,
735            checkpoint_retry_backoff: Duration::ZERO,
736            source_batches_buf: FxHashMap::default(),
737            parked_source_msg: None,
738            source_admission: SourceAdmission::default(),
739            pending_watermark_batches: Vec::new(),
740            barrier_seen: FxHashSet::default(),
741            pending_offsets: vec![None; committed_offsets.len()],
742            replay_pending: false,
743            committed_offsets,
744            control_rx,
745            checkpoint_complete_rx: None,
746            force_ckpt_rx: None,
747            manual_waiting: Vec::new(),
748            manual_handoff_required: false,
749            manual_active: None,
750            checkpoint_in_flight: Arc::new(AtomicU64::new(0)),
751            last_published_checkpoint: None,
752            public_generation: None,
753            intake_stats: Arc::default(),
754            #[cfg(feature = "cluster")]
755            process_authority: source_process_authority,
756        })
757    }
758}
759
760fn source_stage_deadline_error(
761    src_name: &str,
762    start_position: &SourcePosition,
763    error: &str,
764) -> DbError {
765    match start_position {
766                    SourcePosition::Initialized { .. } => DbError::Checkpoint(format!(
767                        "source '{src_name}' sealed initialization failed: {error}"
768                    )),
769                    SourcePosition::Initial => DbError::Config(format!(
770                        "source '{src_name}' start was not attempted: {error}"
771                    )),
772                    SourcePosition::Resume { attempt, .. } => DbError::Checkpoint(format!(
773                        "[LDB-6003] source '{src_name}' start was not attempted while resuming exact checkpoint epoch={} id={}: {error}",
774                        attempt.epoch, attempt.checkpoint_id
775                    )),
776                }
777}
laminar_db/pipeline/streaming_coordinator/construction.rs

laminar_db/pipeline/streaming_coordinator/
construction.rs
