microde_application/
application.rs

1use std::collections::{HashMap, HashSet};
2use std::future::Future;
3use std::panic::AssertUnwindSafe;
4use std::pin::Pin;
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::sync::{Arc, Mutex};
7
8use futures::future::{BoxFuture, FutureExt, join_all, pending, ready};
9use futures::stream::{FuturesUnordered, StreamExt};
10
11#[cfg(test)]
12use crate::MicrodeContext;
13use crate::dependency_graph::DependencyGraph;
14use crate::lifecycle_context::ResolvedRelationship;
15use crate::runtime::{
16    ErrorPriority, ErrorRecorder, InstalledModule, ModuleStage, RuntimeContext, RuntimeControl,
17    spawn, terminate_process,
18};
19use crate::{
20    MicrodeApplicationState, MicrodeContextHandle, MicrodeError, MicrodeExecutionResult,
21    MicrodeModule, MicrodeStopRequest, ModuleFuture, ModuleHandle, ModuleHandleIdentity,
22    ModuleInstanceId, ModuleKind, RelationshipKind, RelationshipSlot, RunContext, SetupContext,
23};
24
25static NEXT_SERVICE_ID: AtomicU64 = AtomicU64::new(0);
26
27#[derive(Clone)]
28struct Binding {
29    owner: ModuleInstanceId,
30    target: ModuleInstanceId,
31    port_id: u64,
32    provider: crate::Provider,
33}
34
35#[derive(Debug, Default, PartialEq, Eq)]
36pub(crate) struct ModuleExecutionErrors {
37    pub(crate) execution: Vec<MicrodeError>,
38    pub(crate) stop: Vec<MicrodeError>,
39}
40
41type ModuleRunFuture =
42    Pin<Box<dyn Future<Output = (usize, ModuleKind, Result<(), MicrodeError>)> + Send + 'static>>;
43
44type ApplicationMain = Box<dyn FnOnce(MicrodeContextHandle) -> ModuleFuture + Send + 'static>;
45
46/// Composes modules and coordinates their lifecycle.
47pub struct MicrodeApplication {
48    pub(crate) modules: Vec<InstalledModule>,
49    pub(crate) context: MicrodeContextHandle,
50    pub(crate) control: Arc<RuntimeControl>,
51    pub(crate) current_state: Arc<Mutex<MicrodeApplicationState>>,
52    composition_id: u64,
53    bindings: HashMap<u64, Binding>,
54    resolutions: HashMap<u64, ResolvedRelationship>,
55    composition_sealed: bool,
56}
57
58struct InstallationStateReset(Arc<Mutex<MicrodeApplicationState>>);
59
60impl Drop for InstallationStateReset {
61    fn drop(&mut self) {
62        *self
63            .0
64            .lock()
65            .unwrap_or_else(std::sync::PoisonError::into_inner) = MicrodeApplicationState::Idle;
66    }
67}
68
69impl MicrodeApplication {
70    /// Creates an idle application with the production module context.
71    pub fn new() -> Self {
72        let control = Arc::new(RuntimeControl::default());
73        let current_state = Arc::new(Mutex::new(MicrodeApplicationState::Idle));
74        let context = Arc::new(RuntimeContext::new(
75            control.clone(),
76            current_state.clone(),
77            terminate_process,
78        ));
79        Self {
80            modules: Vec::new(),
81            context,
82            control,
83            current_state,
84            composition_id: NEXT_SERVICE_ID.fetch_add(1, Ordering::Relaxed),
85            bindings: HashMap::new(),
86            resolutions: HashMap::new(),
87            composition_sealed: false,
88        }
89    }
90
91    #[cfg(test)]
92    pub(crate) fn with_context(context: MicrodeContextHandle) -> Self {
93        Self::with_context_and_control(context, Arc::new(RuntimeControl::default()))
94    }
95
96    #[cfg(test)]
97    pub(crate) fn with_context_and_control(
98        context: MicrodeContextHandle,
99        control: Arc<RuntimeControl>,
100    ) -> Self {
101        Self {
102            modules: Vec::new(),
103            context,
104            control,
105            current_state: Arc::new(Mutex::new(MicrodeApplicationState::Idle)),
106            composition_id: NEXT_SERVICE_ID.fetch_add(1, Ordering::Relaxed),
107            bindings: HashMap::new(),
108            resolutions: HashMap::new(),
109            composition_sealed: false,
110        }
111    }
112
113    pub fn state(&self) -> MicrodeApplicationState {
114        *self
115            .current_state
116            .lock()
117            .unwrap_or_else(std::sync::PoisonError::into_inner)
118    }
119
120    pub fn install<Module, Factory>(&mut self, factory: Factory) -> Result<(), MicrodeError>
121    where
122        Module: MicrodeModule + 'static,
123        Factory: FnOnce(MicrodeContextHandle) -> Module,
124    {
125        match self.ensure_installable() {
126            Ok(()) => {}
127            Err(error) => return Err(error),
128        }
129        self.set_state(MicrodeApplicationState::Installing);
130        let module = {
131            let reset = InstallationStateReset(self.current_state.clone());
132            let module = factory(self.context.clone());
133            drop(reset);
134            module
135        };
136        let id = ModuleInstanceId::new(format!("@installation/{}", self.modules.len()));
137        self.modules.push(InstalledModule::new(id, module));
138        Ok(())
139    }
140
141    pub fn install_named<Module, Factory>(
142        &mut self,
143        id: impl Into<String>,
144        factory: Factory,
145    ) -> Result<ModuleHandle<Module>, MicrodeError>
146    where
147        Module: MicrodeModule + 'static,
148        Factory: FnOnce(MicrodeContextHandle) -> Module,
149    {
150        let id = self.reserve_named_id(id.into())?;
151        self.set_state(MicrodeApplicationState::Installing);
152        let module = {
153            let reset = InstallationStateReset(self.current_state.clone());
154            let module = factory(self.context.clone());
155            drop(reset);
156            module
157        };
158        let handle = ModuleHandle::new(id.clone(), self.composition_id);
159        self.modules.push(InstalledModule::new(id, module));
160        Ok(handle)
161    }
162
163    fn reserve_named_id(&self, value: String) -> Result<ModuleInstanceId, MicrodeError> {
164        self.ensure_installable()?;
165        let id = ModuleInstanceId::new(value);
166        if self.modules.iter().any(|module| module.id() == &id) {
167            return Err(MicrodeError::new(format!(
168                "module instance ID '{}' is already installed",
169                id.as_str()
170            )));
171        }
172        Ok(id)
173    }
174
175    pub fn bind(
176        &mut self,
177        consumer: &dyn ModuleHandleIdentity,
178        slot: &dyn RelationshipSlot,
179        target: &dyn ModuleHandleIdentity,
180    ) -> Result<(), MicrodeError> {
181        self.ensure_installable()?;
182        for handle in [
183            (consumer.module_instance_id(), consumer.composition_owner()),
184            (target.module_instance_id(), target.composition_owner()),
185        ] {
186            if handle.1 != self.composition_id {
187                return Err(MicrodeError::new(format!(
188                    "module handle '{}' belongs to another application",
189                    handle.0.as_str()
190                )));
191            }
192        }
193        let descriptor = slot.descriptor();
194        let installed = self
195            .modules
196            .iter()
197            .find(|module| module.id() == consumer.module_instance_id())
198            .unwrap();
199        if !installed
200            .relationships()
201            .iter()
202            .any(|known| known.slot_id == descriptor.slot_id)
203        {
204            return Err(MicrodeError::new(format!(
205                "unknown relationship '{}.{}'",
206                consumer.module_instance_id().as_str(),
207                descriptor.name
208            )));
209        }
210        if self.bindings.contains_key(&descriptor.slot_id) {
211            return Err(MicrodeError::new(format!(
212                "relationship '{}.{}' is already bound",
213                consumer.module_instance_id().as_str(),
214                descriptor.name
215            )));
216        }
217        let provider = self
218            .modules
219            .iter()
220            .find(|module| module.id() == target.module_instance_id())
221            .unwrap();
222        if let Some((required, name)) = descriptor.module_type
223            && provider.module_type() != required
224        {
225            return Err(MicrodeError::new(format!(
226                "module '{}' does not satisfy concrete module requirement '{}'",
227                target.module_instance_id().as_str(),
228                name.rsplit("::").next().unwrap_or(name)
229            )));
230        }
231        let Some(exported) = provider
232            .providers()
233            .iter()
234            .find(|known| known.port_id == descriptor.port_id)
235        else {
236            return Err(MicrodeError::new(format!(
237                "module '{}' does not provide port '{}'",
238                target.module_instance_id().as_str(),
239                descriptor.port_description
240            )));
241        };
242        self.bindings.insert(
243            descriptor.slot_id,
244            Binding {
245                owner: consumer.module_instance_id().clone(),
246                target: target.module_instance_id().clone(),
247                port_id: descriptor.port_id,
248                provider: exported.clone(),
249            },
250        );
251        Ok(())
252    }
253
254    fn wire_composition(&mut self) -> Result<(), MicrodeError> {
255        let mut graph = DependencyGraph::new(
256            self.modules
257                .iter()
258                .map(|module| module.id().clone())
259                .collect(),
260        );
261        for module in &self.modules {
262            for relationship in module.relationships() {
263                let binding = self.bindings.get(&relationship.slot_id).ok_or_else(|| {
264                    MicrodeError::new(format!(
265                        "missing binding for relationship '{}.{}'",
266                        module.id().as_str(),
267                        relationship.name
268                    ))
269                })?;
270                if relationship.kind == RelationshipKind::Dependency {
271                    graph.add_validated_dependency(&binding.owner, &binding.target);
272                }
273            }
274        }
275        let order = graph.order()?;
276        let mut staged = HashMap::new();
277        let mut provider_values: HashMap<
278            (ModuleInstanceId, u64),
279            Arc<dyn std::any::Any + Send + Sync>,
280        > = HashMap::new();
281        for module in &self.modules {
282            for relationship in module.relationships() {
283                let binding = &self.bindings[&relationship.slot_id];
284                let provider_key = (binding.target.clone(), binding.port_id);
285                let value = match provider_values.get(&provider_key) {
286                    Some(value) => value.clone(),
287                    None => {
288                        let value = binding.provider.resolve()?;
289                        provider_values.insert(provider_key, value.clone());
290                        value
291                    }
292                };
293                staged.insert(
294                    relationship.slot_id,
295                    ResolvedRelationship {
296                        owner: module.id().clone(),
297                        name: relationship.name.clone(),
298                        kind: relationship.kind,
299                        value,
300                    },
301                );
302            }
303        }
304        self.modules
305            .sort_by_key(|module| order.iter().position(|id| id == module.id()));
306        self.resolutions = staged;
307        Ok(())
308    }
309
310    fn ensure_installable(&self) -> Result<(), MicrodeError> {
311        if self.state() != MicrodeApplicationState::Idle {
312            return Err(MicrodeError::new(format!(
313                "cannot install module after application has started; current state: {:?}",
314                self.state()
315            )));
316        }
317        if self.composition_sealed {
318            return Err(MicrodeError::new(
319                "cannot modify composition after it is sealed",
320            ));
321        }
322        Ok(())
323    }
324
325    pub(crate) fn set_state(&self, state: MicrodeApplicationState) {
326        *self
327            .current_state
328            .lock()
329            .unwrap_or_else(std::sync::PoisonError::into_inner) = state;
330    }
331
332    /// Serves the application using module completion and stop requests to control its lifetime.
333    ///
334    /// The lifecycle continues if the returned future is dropped. Any later call to [`Self::stop`]
335    /// receives the same shared completion result.
336    pub fn serve(&mut self) -> BoxFuture<'static, Result<MicrodeExecutionResult, MicrodeError>> {
337        self.start(None)
338    }
339
340    /// Runs an application-level task after all modules have started.
341    ///
342    /// Completion or failure of the task begins orderly application shutdown.
343    pub fn run<Main, MainFuture>(
344        &mut self,
345        main: Main,
346    ) -> BoxFuture<'static, Result<MicrodeExecutionResult, MicrodeError>>
347    where
348        Main: FnOnce(MicrodeContextHandle) -> MainFuture + Send + 'static,
349        MainFuture: Future<Output = Result<(), MicrodeError>> + Send + 'static,
350    {
351        self.start(Some(Box::new(move |context| Box::pin(main(context)))))
352    }
353
354    fn start(
355        &mut self,
356        main: Option<ApplicationMain>,
357    ) -> BoxFuture<'static, Result<MicrodeExecutionResult, MicrodeError>> {
358        if self.state() != MicrodeApplicationState::Idle {
359            return ready(Err(MicrodeError::new(format!(
360                "cannot start application more than once; current state: {:?}",
361                self.state()
362            ))))
363            .boxed();
364        }
365
366        if self.composition_sealed {
367            return ready(Err(MicrodeError::new(
368                "cannot start application more than once; composition is sealed",
369            )))
370            .boxed();
371        }
372        self.composition_sealed = true;
373
374        if let Err(error) = self.wire_composition() {
375            return ready(Err(error)).boxed();
376        }
377
378        self.set_state(MicrodeApplicationState::Initialization);
379        let mut runner = Self {
380            modules: std::mem::take(&mut self.modules),
381            context: self.context.clone(),
382            control: self.control.clone(),
383            current_state: self.current_state.clone(),
384            composition_id: self.composition_id,
385            bindings: std::mem::take(&mut self.bindings),
386            resolutions: std::mem::take(&mut self.resolutions),
387            composition_sealed: true,
388        };
389
390        let control = runner.control.clone();
391        let completion = control.clone();
392        spawn(async move {
393            let result = AssertUnwindSafe(runner.execute_lifecycle(main))
394                .catch_unwind()
395                .await
396                .map_err(|panic| {
397                    runner.set_state(MicrodeApplicationState::Failed);
398                    MicrodeError::new(panic_message(panic))
399                });
400            control.complete(result);
401        });
402
403        async move { completion.wait_for_completion().await }.boxed()
404    }
405
406    /// Requests an orderly stop and waits for lifecycle completion.
407    pub fn stop(
408        &self,
409        request: MicrodeStopRequest,
410    ) -> BoxFuture<'static, Result<MicrodeExecutionResult, MicrodeError>> {
411        let state = self.state();
412        if matches!(
413            state,
414            MicrodeApplicationState::Idle | MicrodeApplicationState::Installing
415        ) {
416            return ready(Err(MicrodeError::new(format!(
417                "cannot stop application before it has started; current state: {state:?}"
418            ))))
419            .boxed();
420        }
421
422        self.control.request_stop(request);
423        let control = self.control.clone();
424        async move { control.wait_for_completion().await }.boxed()
425    }
426
427    async fn execute_lifecycle(&mut self, main: Option<ApplicationMain>) -> MicrodeExecutionResult {
428        let mut errors = ErrorRecorder::default();
429        let mut forward_failed = false;
430
431        if let Err(error) = self.initialize_modules().await {
432            errors.record(error, ErrorPriority::Lifecycle);
433            forward_failed = true;
434        }
435
436        if !forward_failed && !self.control.stop_requested() {
437            self.set_state(MicrodeApplicationState::Setup);
438            if let Err(error) = self.setup_modules().await {
439                errors.record(error, ErrorPriority::Lifecycle);
440                forward_failed = true;
441            }
442        }
443
444        if !forward_failed && !self.control.stop_requested() {
445            self.set_state(MicrodeApplicationState::Running);
446            let execution_errors = self.execute_modules(main).await;
447            for error in execution_errors.execution {
448                errors.record(error, ErrorPriority::Execution);
449            }
450            for error in execution_errors.stop {
451                errors.record(error, ErrorPriority::Stop);
452            }
453        }
454
455        self.set_state(MicrodeApplicationState::TearDown);
456        for error in self.teardown_modules().await {
457            errors.record(error, ErrorPriority::Lifecycle);
458        }
459
460        self.set_state(MicrodeApplicationState::Shutdown);
461        for error in self.shutdown_modules().await {
462            errors.record(error, ErrorPriority::Lifecycle);
463        }
464
465        self.set_state(MicrodeApplicationState::CleanUp);
466        for error in self.cleanup_modules().await {
467            errors.record(error, ErrorPriority::Lifecycle);
468        }
469
470        let stop_request = self.control.stop_request();
471        let exit_code = stop_request.as_ref().and_then(|request| request.exit_code);
472        if let Some(error) = stop_request.and_then(|request| request.error) {
473            errors.record(error, ErrorPriority::StopRequest);
474        }
475
476        let result = errors.into_result(exit_code);
477        if result.error.is_some() || result.exit_code != 0 {
478            self.set_state(MicrodeApplicationState::Failed);
479        } else {
480            self.set_state(MicrodeApplicationState::Finished);
481        }
482        result
483    }
484
485    pub(crate) async fn initialize_modules(&mut self) -> Result<(), MicrodeError> {
486        for installed in &mut self.modules {
487            if self.control.stop_requested() {
488                break;
489            }
490
491            installed.set_stage(ModuleStage::Initializing);
492            match installed.initialize().await {
493                Ok(()) => installed.set_stage(ModuleStage::Initialized),
494                Err(error) => return Err(error),
495            }
496        }
497        Ok(())
498    }
499
500    pub(crate) async fn setup_modules(&mut self) -> Result<(), MicrodeError> {
501        let resolutions = Arc::new(self.resolutions.clone());
502        for installed in &mut self.modules {
503            if self.control.stop_requested() {
504                break;
505            }
506
507            installed.set_stage(ModuleStage::SettingUp);
508            let context = SetupContext::new(installed.id().clone(), resolutions.clone());
509            match installed.setup_with_context(context).await {
510                Ok(()) => installed.set_stage(ModuleStage::SetUp),
511                Err(error) => return Err(error),
512            }
513        }
514        Ok(())
515    }
516
517    pub(crate) async fn execute_modules(
518        &mut self,
519        main: Option<ApplicationMain>,
520    ) -> ModuleExecutionErrors {
521        let mut errors = ModuleExecutionErrors::default();
522        let mut runs: FuturesUnordered<ModuleRunFuture> = FuturesUnordered::new();
523        let resolutions = Arc::new(self.resolutions.clone());
524
525        for (index, installed) in self.modules.iter_mut().enumerate() {
526            installed.set_stage(ModuleStage::Executing);
527            let kind = installed.kind();
528            let context = RunContext::new(installed.id().clone(), resolutions.clone());
529            let run = installed.run_with_context(context);
530            runs.push(Box::pin(async move { (index, kind, run.await) }));
531        }
532
533        let stop_signal = self.control.take_stop_receiver().fuse();
534        futures::pin_mut!(stop_signal);
535
536        let has_main = main.is_some();
537        let main_future = match main {
538            Some(main) => main(self.context.clone()).boxed(),
539            None => pending::<Result<(), MicrodeError>>().boxed(),
540        }
541        .fuse();
542        futures::pin_mut!(main_future);
543
544        loop {
545            if runs.is_empty() {
546                if !has_main {
547                    break;
548                }
549                futures::select_biased! {
550                    _ = stop_signal => break,
551                    result = main_future => {
552                        if let Err(error) = result {
553                            errors.execution.push(error);
554                        }
555                        break;
556                    }
557                }
558            }
559            futures::select_biased! {
560                _ = stop_signal => break,
561                result = main_future => {
562                    if let Err(error) = result {
563                        errors.execution.push(error);
564                    }
565                    break;
566                },
567                outcome = runs.next().fuse() => {
568                    let (index, kind, result) = outcome
569                        .expect("non-empty module runs have a next completion");
570                    let should_stop = kind == ModuleKind::Active || result.is_err();
571                    self.record_run_completion((index, kind, result), &mut errors);
572                    if should_stop {
573                        break;
574                    }
575                }
576            }
577        }
578
579        let module_indexes = self
580            .modules
581            .iter()
582            .enumerate()
583            .rev()
584            .map(|(index, _)| index)
585            .collect::<Vec<_>>();
586        let mut stop_futures = Vec::with_capacity(module_indexes.len());
587        for index in module_indexes {
588            let stop = self.modules[index].stop();
589            stop_futures.push(async move { (index, stop.await) });
590        }
591
592        let mut required_completions = self
593            .modules
594            .iter()
595            .enumerate()
596            .filter_map(|(index, module)| {
597                (module.kind() == ModuleKind::Passive && module.stage() != ModuleStage::Executed)
598                    .then_some(index)
599            })
600            .collect::<HashSet<_>>();
601
602        for (index, result) in join_all(stop_futures).await {
603            match result {
604                Ok(()) => {
605                    if self.modules[index].stage() != ModuleStage::Executed {
606                        required_completions.insert(index);
607                    }
608                }
609                Err(error) => errors.stop.push(error),
610            }
611        }
612
613        while !required_completions.is_empty() {
614            let outcome = runs
615                .next()
616                .await
617                .expect("required module completions have corresponding run futures");
618            required_completions.remove(&outcome.0);
619            self.record_run_completion(outcome, &mut errors);
620        }
621
622        // JavaScript promises continue running even when the lifecycle no longer awaits them.
623        // Preserve that behavior for active runs whose stop operation failed instead of
624        // cancelling their futures when `runs` is dropped.
625        if !runs.is_empty() {
626            spawn(async move { while runs.next().await.is_some() {} });
627        }
628
629        errors
630    }
631
632    fn record_run_completion(
633        &mut self,
634        (index, _kind, result): (usize, ModuleKind, Result<(), MicrodeError>),
635        errors: &mut ModuleExecutionErrors,
636    ) {
637        self.modules[index].set_stage(ModuleStage::Executed);
638        if let Err(error) = result {
639            errors.execution.push(error);
640        }
641    }
642
643    pub(crate) async fn teardown_modules(&mut self) -> Vec<MicrodeError> {
644        let mut errors = Vec::new();
645        for installed in self.modules.iter_mut().rev() {
646            if installed.stage() < ModuleStage::SettingUp
647                || installed.stage() >= ModuleStage::TearingDown
648            {
649                continue;
650            }
651
652            installed.set_stage(ModuleStage::TearingDown);
653            if let Err(error) = installed.teardown().await {
654                errors.push(error);
655            }
656            installed.set_stage(ModuleStage::TornDown);
657        }
658        errors
659    }
660
661    pub(crate) async fn shutdown_modules(&mut self) -> Vec<MicrodeError> {
662        let mut errors = Vec::new();
663        for installed in self.modules.iter_mut().rev() {
664            if installed.stage() < ModuleStage::Initializing
665                || installed.stage() >= ModuleStage::ShuttingDown
666            {
667                continue;
668            }
669
670            installed.set_stage(ModuleStage::ShuttingDown);
671            if let Err(error) = installed.shutdown().await {
672                errors.push(error);
673            }
674            installed.set_stage(ModuleStage::Shutdown);
675        }
676        errors
677    }
678
679    pub(crate) async fn cleanup_modules(&mut self) -> Vec<MicrodeError> {
680        let mut errors = Vec::new();
681        for installed in self.modules.iter_mut().rev() {
682            installed.set_stage(ModuleStage::CleaningUp);
683            if let Err(error) = installed.cleanup().await {
684                errors.push(error);
685            }
686            installed.set_stage(ModuleStage::CleanedUp);
687        }
688        errors
689    }
690}
691
692impl Default for MicrodeApplication {
693    fn default() -> Self {
694        Self::new()
695    }
696}
697
698fn panic_message(panic: Box<dyn std::any::Any + Send>) -> String {
699    if let Some(message) = panic.downcast_ref::<&str>() {
700        return (*message).to_owned();
701    }
702    if let Some(message) = panic.downcast_ref::<String>() {
703        return message.clone();
704    }
705    "application lifecycle panicked".to_owned()
706}
707
708#[cfg(test)]
709#[path = "tests/composition_wiring.rs"]
710mod composition_wiring_tests;
711#[cfg(test)]
712#[path = "tests/dependency_lifecycle.rs"]
713mod dependency_lifecycle_tests;
714#[cfg(test)]
715#[path = "tests/execution.rs"]
716mod execution_tests;
717#[cfg(test)]
718#[path = "tests/initialization_and_setup.rs"]
719mod initialization_and_setup_tests;
720#[cfg(test)]
721#[path = "tests/installation.rs"]
722mod installation_tests;
723#[cfg(test)]
724#[path = "tests/public_runtime.rs"]
725mod public_runtime_tests;
726#[cfg(test)]
727#[path = "tests/unwind.rs"]
728mod unwind_tests;
microde_application/application.rs

microde_application/
application.rs
