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

laminar_db/pipeline/streaming_coordinator/
construction.rs
