ri/core/app_runtime.rs
1//! Copyright © 2025-2026 Wenze Wei. All Rights Reserved.
2//!
3//! This file is part of Ri.
4//! The Ri project belongs to the Dunimd Team.
5//!
6//! Licensed under the Apache License, Version 2.0 (the "License");
7//! You may not use this file except in compliance with the License.
8//! You may obtain a copy of the License at
9//!
10//! http://www.apache.org/licenses/LICENSE-2.0
11//!
12//! Unless required by applicable law or agreed to in writing, software
13//! distributed under the License is distributed on an "AS IS" BASIS,
14//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15//! See the License for the specific language governing permissions and
16//! limitations under the License.
17
18#![allow(non_snake_case)]
19
20//! # Application Runtime
21//!
22//! This module provides the application runtime for Ri applications.
23//! The `RiAppRuntime` manages the application lifecycle, including module initialization,
24//! startup, and shutdown. It also handles the execution of both synchronous and asynchronous modules.
25
26use crate::core::{RiResult, RiServiceContext};
27use crate::hooks::{RiHookKind, RiModulePhase};
28use super::module_types::{ModuleSlot, ModuleType};
29use tokio::sync::RwLock as AsyncRwLock;
30use std::sync::Arc;
31use std::sync::atomic::{AtomicBool, Ordering};
32#[cfg(feature = "pyo3")]
33use pyo3::prelude::*;
34
35/// Public-facing application runtime.
36///
37/// The `RiAppRuntime` manages the application lifecycle, including module initialization,
38/// startup, and shutdown. It also handles the execution of both synchronous and asynchronous modules.
39///
40/// ## Usage
41///
42/// ```rust
43/// use ri::prelude::*;
44///
45/// #[tokio::main]
46/// async fn main() -> RiResult<()> {
47/// let app = RiAppBuilder::new()
48/// .with_config("config.yaml")?
49/// .build()?;
50///
51/// app.run(|ctx| async move {
52/// ctx.logger().info("service", "Ri service started")?;
53/// Ok(())
54/// }).await
55/// }
56/// ```
57
58#[cfg_attr(feature = "pyo3", pyo3::prelude::pyclass)]
59#[derive(Clone)]
60pub struct RiAppRuntime {
61 /// Service context providing access to core functionalities
62 ctx: RiServiceContext,
63 /// Vector of modules with their state, protected by an async RwLock
64 modules: Arc<AsyncRwLock<Vec<ModuleSlot>>>,
65 /// Whether the application lifecycle is currently running
66 running: Arc<AtomicBool>,
67}
68
69impl RiAppRuntime {
70 /// Create a new application runtime with the given context and modules.
71 ///
72 /// This method is typically called by the `RiAppBuilder` during the build process.
73 ///
74 /// # Parameters
75 ///
76 /// - `ctx`: Service context with core functionalities
77 /// - `modules`: Vector of modules with their initial state
78 ///
79 /// # Returns
80 ///
81 /// A new `RiAppRuntime` instance.
82 pub fn new(ctx: RiServiceContext, modules: Vec<ModuleSlot>) -> Self {
83 Self {
84 ctx,
85 modules: Arc::new(AsyncRwLock::new(modules)),
86 running: Arc::new(AtomicBool::new(false)),
87 }
88 }
89
90 /// Check whether the application lifecycle is currently running.
91 ///
92 /// This returns `true` between the start of `run` and the completion of the
93 /// shutdown sequence (including when `run` returns early with an error).
94 pub fn is_running(&self) -> bool {
95 self.running.load(Ordering::SeqCst)
96 }
97
98 /// Request a cooperative shutdown of the application.
99 ///
100 /// This method is idempotent: if the lifecycle is not running (either it has
101 /// not started yet or it already finished), it returns `Ok(())` immediately.
102 /// Otherwise it executes the module shutdown sequence (sync modules first in
103 /// reverse order, then async modules in reverse order, then shutdown hooks)
104 /// and marks the lifecycle as no longer running.
105 ///
106 /// # Errors
107 ///
108 /// Returns an error if a critical module fails during shutdown.
109 pub async fn shutdown(&mut self) -> RiResult<()> {
110 if !self.running.swap(false, Ordering::SeqCst) {
111 return Ok(());
112 }
113 self.shutdown_all_modules().await
114 }
115
116 /// Run the application lifecycle.
117 ///
118 /// This method executes the complete application lifecycle, including:
119 /// 1. Emitting startup hooks
120 /// 2. Initializing synchronous modules
121 /// 3. Starting synchronous modules
122 /// 4. Initializing and starting asynchronous modules
123 /// 5. Running the application business logic via the provided closure
124 /// 6. Shutting down asynchronous modules
125 /// 7. Shutting down synchronous modules
126 /// 8. Emitting shutdown hooks
127 ///
128 /// # Parameters
129 ///
130 /// - `f`: A closure that takes a `RiServiceContext` and returns a `RiResult<()>`.
131 /// This closure contains the application's business logic and is executed after all
132 /// modules have been initialized and started, but before any modules are shut down.
133 ///
134 /// # Returns
135 ///
136 /// A `RiResult` indicating success or failure.
137 ///
138 /// # Errors
139 ///
140 /// Returns an error if:
141 /// - A critical module fails during execution
142 /// - The provided closure returns an error
143 pub async fn run<F, Fut>(mut self, f: F) -> RiResult<()>
144 where
145 F: FnOnce(&RiServiceContext) -> Fut,
146 Fut: std::future::Future<Output = RiResult<()>>,
147 {
148 self.running.store(true, Ordering::SeqCst);
149 let result = self.run_inner(f).await;
150 self.running.store(false, Ordering::SeqCst);
151 result
152 }
153
154 /// Internal implementation of the application lifecycle.
155 async fn run_inner<F, Fut>(&mut self, f: F) -> RiResult<()>
156 where
157 F: FnOnce(&RiServiceContext) -> Fut,
158 Fut: std::future::Future<Output = RiResult<()>>,
159 {
160 // Emit startup hook
161 self.ctx.hooks().emit_with(&RiHookKind::Startup, &self.ctx, None, None)?;
162
163 // Emit before modules init hook
164 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesInit, &self.ctx, None, None)?;
165
166 // Get module count
167 let modules_guard = self.modules.read().await;
168 let module_len = modules_guard.len();
169 drop(modules_guard); // Release lock early
170
171 // Collect module states first to avoid repeated lock acquisitions
172 let mut module_states = Vec::with_capacity(module_len);
173 {
174 let modules_guard = self.modules.read().await;
175 for idx in 0..module_len {
176 if idx < modules_guard.len() {
177 let slot = &modules_guard[idx];
178 module_states.push((
179 idx,
180 !slot.failed,
181 if !slot.failed {
182 slot.module.name().to_string()
183 } else {
184 String::new()
185 },
186 if !slot.failed {
187 slot.module.is_critical()
188 } else {
189 false
190 },
191 ));
192 } else {
193 module_states.push((idx, false, String::new(), false));
194 }
195 }
196 }
197
198 // Initialize synchronous modules
199 for (idx, skip, module_name, critical) in module_states.iter().cloned() {
200 if !skip {
201 // Emit before module init hook
202 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesInit, &self.ctx, Some(&module_name), Some(RiModulePhase::Init))?;
203
204 // Initialize module with single write lock acquisition
205 let mut error = None;
206 {
207 let mut modules_guard = self.modules.write().await;
208 if idx < modules_guard.len() {
209 match &mut modules_guard[idx].module {
210 ModuleType::Sync(_module) => {
211 if let Err(err) = _module.init(&mut self.ctx) {
212 error = Some(err);
213 }
214 }
215 ModuleType::Async(_module) => {
216 // Async modules are handled separately in the async phase
217 }
218 }
219 }
220 }
221
222 // Handle module initialization error
223 if let Some(err) = error {
224 self.log_module_error("init", &module_name, &err);
225 if critical {
226 return Err(err);
227 } else {
228 // Mark module as failed with single write lock acquisition
229 let mut modules_guard = self.modules.write().await;
230 if idx < modules_guard.len() {
231 modules_guard[idx].failed = true;
232 }
233 }
234 }
235 }
236 }
237
238 // Emit after modules init hook
239 self.ctx.hooks().emit_with(&RiHookKind::AfterModulesInit, &self.ctx, None, None)?;
240
241 // Emit before modules start hook
242 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesStart, &self.ctx, None, None)?;
243
244 // Start synchronous modules with optimized locking
245 for (idx, skip, module_name, critical) in module_states.iter().cloned() {
246 if !skip {
247 let mut err_phase = "start";
248
249 // Execute all sync module phases with single write lock acquisition
250 let mut error = None;
251 {
252 let mut modules_guard = self.modules.write().await;
253 if idx < modules_guard.len() {
254 match &mut modules_guard[idx].module {
255 ModuleType::Sync(_module) => {
256 // Execute before_start phase
257 if let Err(err) = _module.before_start(&mut self.ctx) {
258 err_phase = "before_start";
259 error = Some(err);
260 }
261
262 // Execute start phase if no error
263 if error.is_none() {
264 if let Err(err) = _module.start(&mut self.ctx) {
265 err_phase = "start";
266 error = Some(err);
267 }
268 }
269
270 // Execute after_start phase if no error
271 if error.is_none() {
272 if let Err(err) = _module.after_start(&mut self.ctx) {
273 err_phase = "after_start";
274 error = Some(err);
275 }
276 }
277 }
278 ModuleType::Async(_module) => {
279 // Async modules are handled separately in the async phase
280 }
281 }
282 }
283 }
284
285 // Emit hooks outside of lock to avoid potential deadlocks
286 if error.is_none() {
287 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesStart, &self.ctx, Some(&module_name), Some(RiModulePhase::BeforeStart))?;
288 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesStart, &self.ctx, Some(&module_name), Some(RiModulePhase::Start))?;
289 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesStart, &self.ctx, Some(&module_name), Some(RiModulePhase::AfterStart))?;
290 }
291
292 // Handle module start error
293 if let Some(err) = error {
294 self.log_module_error(err_phase, &module_name, &err);
295 if critical {
296 return Err(err);
297 } else {
298 // Mark module as failed with single write lock acquisition
299 let mut modules_guard = self.modules.write().await;
300 if idx < modules_guard.len() {
301 modules_guard[idx].failed = true;
302 }
303 }
304 }
305 }
306 }
307
308 // Emit after modules start hook
309 self.ctx.hooks().emit_with(&RiHookKind::AfterModulesStart, &self.ctx, None, None)?;
310
311 // Initialize and start asynchronous modules with optimized locking
312 for idx in 0..module_len {
313 let mut err_phase = "async_start";
314
315 // Check if this is an async module and get its state
316 let (skip, module_name, critical) = {
317 let modules_guard = self.modules.read().await;
318 if idx < modules_guard.len() {
319 let slot = &modules_guard[idx];
320 if !slot.failed {
321 match &slot.module {
322 ModuleType::Async(module) => (
323 false,
324 module.name().to_string(),
325 module.is_critical(),
326 ),
327 ModuleType::Sync(_) => (true, String::new(), false),
328 }
329 } else {
330 (true, String::new(), false)
331 }
332 } else {
333 (true, String::new(), false)
334 }
335 };
336
337 if !skip {
338 // Execute all async module phases with single write lock acquisition
339 let mut error = None;
340 {
341 let mut modules_guard = self.modules.write().await;
342 if idx < modules_guard.len() {
343 if let ModuleType::Async(_module) = &mut modules_guard[idx].module {
344 // Execute async init phase
345 if let Err(err) = _module.init(&mut self.ctx).await {
346 err_phase = "async_init";
347 error = Some(err);
348 }
349
350 // Execute async before_start phase if no error
351 if error.is_none() {
352 if let Err(err) = _module.before_start(&mut self.ctx).await {
353 err_phase = "async_before_start";
354 error = Some(err);
355 }
356 }
357
358 // Execute async start phase if no error
359 if error.is_none() {
360 if let Err(err) = _module.start(&mut self.ctx).await {
361 err_phase = "async_start";
362 error = Some(err);
363 }
364 }
365
366 // Execute async after_start phase if no error
367 if error.is_none() {
368 if let Err(err) = _module.after_start(&mut self.ctx).await {
369 err_phase = "async_after_start";
370 error = Some(err);
371 }
372 }
373 }
374 }
375 }
376
377 // Emit hooks outside of lock to avoid potential deadlocks
378 if error.is_none() {
379 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesStart, &self.ctx, Some(&module_name), Some(RiModulePhase::AsyncInit))?;
380 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesStart, &self.ctx, Some(&module_name), Some(RiModulePhase::AsyncBeforeStart))?;
381 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesStart, &self.ctx, Some(&module_name), Some(RiModulePhase::AsyncStart))?;
382 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesStart, &self.ctx, Some(&module_name), Some(RiModulePhase::AsyncAfterStart))?;
383 }
384
385 // Handle async module error
386 if let Some(err) = error {
387 self.log_module_error(err_phase, &module_name, &err);
388 if critical {
389 return Err(err);
390 } else {
391 // Mark module as failed with single write lock acquisition
392 let mut modules_guard = self.modules.write().await;
393 if idx < modules_guard.len() {
394 modules_guard[idx].failed = true;
395 }
396 }
397 }
398 }
399 }
400
401 // Emit after async modules start hook
402 self.ctx.hooks().emit_with(&RiHookKind::AfterModulesStart, &self.ctx, None, None)?;
403
404 // Run the application business logic (provided closure)
405 let result = f(&self.ctx).await;
406
407 self.shutdown_all_modules().await?;
408
409 // Return the result of the closure execution
410 result
411 }
412
413 /// Execute the module shutdown sequence.
414 ///
415 /// This shuts down synchronous modules in reverse order, then asynchronous
416 /// modules in reverse order, emitting the corresponding lifecycle hooks.
417 /// The `running` flag is not touched here; callers manage it.
418 ///
419 /// # Errors
420 ///
421 /// Returns an error if hook emission fails or a critical module fails during
422 /// shutdown. Non-critical failures mark the module slot as failed and are
423 /// logged instead of aborting the sequence.
424 async fn shutdown_all_modules(&mut self) -> RiResult<()> {
425 let modules_guard = self.modules.read().await;
426 let module_len = modules_guard.len();
427 drop(modules_guard);
428
429 // Emit before modules shutdown hook
430 // Note: We're using a new context here since we've moved the original to the closure
431 let _ = self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesShutdown, &self.ctx, None, None);
432
433 // Shutdown synchronous modules in reverse order with optimized locking
434 for idx in (0..module_len).rev() {
435 // Check if this is a sync module and get its state
436 let (skip, module_name, critical) = {
437 let modules_guard = self.modules.read().await;
438 if idx < modules_guard.len() {
439 let slot = &modules_guard[idx];
440 if !slot.failed {
441 match &slot.module {
442 ModuleType::Sync(module) => (
443 false,
444 module.name().to_string(),
445 module.is_critical(),
446 ),
447 ModuleType::Async(_) => (true, String::new(), false),
448 }
449 } else {
450 (true, String::new(), false)
451 }
452 } else {
453 (true, String::new(), false)
454 }
455 };
456
457 if !skip {
458 // Execute all sync module shutdown phases with single write lock acquisition
459 let mut err_phase = "shutdown";
460 let mut error = None;
461 {
462 let mut modules_guard = self.modules.write().await;
463 if idx < modules_guard.len() {
464 if let ModuleType::Sync(_module) = &mut modules_guard[idx].module {
465 // Execute before_shutdown phase
466 if let Err(err) = _module.before_shutdown(&mut self.ctx) {
467 err_phase = "before_shutdown";
468 error = Some(err);
469 }
470
471 // Execute shutdown phase if no error
472 if error.is_none() {
473 if let Err(err) = _module.shutdown(&mut self.ctx) {
474 err_phase = "shutdown";
475 error = Some(err);
476 }
477 }
478
479 // Execute after_shutdown phase if no error
480 if error.is_none() {
481 if let Err(err) = _module.after_shutdown(&mut self.ctx) {
482 err_phase = "after_shutdown";
483 error = Some(err);
484 }
485 }
486 }
487 }
488 }
489
490 // Emit hooks outside of lock to avoid potential deadlocks
491 if error.is_none() {
492 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesShutdown, &self.ctx, Some(&module_name), Some(RiModulePhase::BeforeShutdown))?;
493 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesShutdown, &self.ctx, Some(&module_name), Some(RiModulePhase::Shutdown))?;
494 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesShutdown, &self.ctx, Some(&module_name), Some(RiModulePhase::AfterShutdown))?;
495 }
496
497 // Handle module shutdown error
498 if let Some(err) = error {
499 self.log_module_error(err_phase, &module_name, &err);
500 if critical {
501 return Err(err);
502 } else {
503 // Mark module as failed
504 let mut modules_guard = self.modules.write().await;
505 if idx < modules_guard.len() {
506 modules_guard[idx].failed = true;
507 }
508 }
509 }
510 }
511 }
512
513 // Emit after modules shutdown hook
514 self.ctx.hooks().emit_with(&RiHookKind::AfterModulesShutdown, &self.ctx, None, None)?;
515
516 // Shutdown asynchronous modules in reverse order with optimized locking
517 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesShutdown, &self.ctx, None, None)?;
518
519 for idx in (0..module_len).rev() {
520 // Check if this is an async module and get its state
521 let (skip, module_name, critical) = {
522 let modules_guard = self.modules.read().await;
523 if idx < modules_guard.len() {
524 let slot = &modules_guard[idx];
525 if !slot.failed {
526 match &slot.module {
527 ModuleType::Async(module) => (
528 false,
529 module.name().to_string(),
530 module.is_critical(),
531 ),
532 ModuleType::Sync(_) => (true, String::new(), false),
533 }
534 } else {
535 (true, String::new(), false)
536 }
537 } else {
538 (true, String::new(), false)
539 }
540 };
541
542 if !skip {
543 // Execute all async module shutdown phases with single write lock acquisition
544 let mut err_phase = "async_shutdown";
545 let mut error = None;
546 {
547 let mut modules_guard = self.modules.write().await;
548 if idx < modules_guard.len() {
549 if let ModuleType::Async(_module) = &mut modules_guard[idx].module {
550 // Execute async before_shutdown phase
551 if let Err(err) = _module.before_shutdown(&mut self.ctx).await {
552 err_phase = "async_before_shutdown";
553 error = Some(err);
554 }
555
556 // Execute async shutdown phase if no error
557 if error.is_none() {
558 if let Err(err) = _module.shutdown(&mut self.ctx).await {
559 err_phase = "async_shutdown";
560 error = Some(err);
561 }
562 }
563
564 // Execute async after_shutdown phase if no error
565 if error.is_none() {
566 if let Err(err) = _module.after_shutdown(&mut self.ctx).await {
567 err_phase = "async_after_shutdown";
568 error = Some(err);
569 }
570 }
571 }
572 }
573 }
574
575 // Emit hooks outside of lock to avoid potential deadlocks
576 if error.is_none() {
577 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesShutdown, &self.ctx, Some(&module_name), Some(RiModulePhase::AsyncBeforeShutdown))?;
578 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesShutdown, &self.ctx, Some(&module_name), Some(RiModulePhase::AsyncShutdown))?;
579 self.ctx.hooks().emit_with(&RiHookKind::BeforeModulesShutdown, &self.ctx, Some(&module_name), Some(RiModulePhase::AsyncAfterShutdown))?;
580 }
581
582 // Handle async module shutdown error
583 if let Some(err) = error {
584 self.log_module_error(err_phase, &module_name, &err);
585 if critical {
586 return Err(err);
587 } else {
588 // Mark module as failed with single write lock acquisition
589 let mut modules_guard = self.modules.write().await;
590 if idx < modules_guard.len() {
591 modules_guard[idx].failed = true;
592 }
593 }
594 }
595 }
596 }
597
598 // Emit after async modules shutdown hook
599 self.ctx.hooks().emit_with(&RiHookKind::AfterModulesShutdown, &self.ctx, None, None)?;
600
601 // Emit shutdown hook
602 self.ctx.hooks().emit_with(&RiHookKind::Shutdown, &self.ctx, None, None)?;
603
604 Ok(())
605 }
606
607 /// Log a module error.
608 ///
609 /// This method logs an error that occurred during module execution, including
610 /// the module name, phase, and error message.
611 ///
612 /// # Parameters
613 ///
614 /// - `phase`: The lifecycle phase during which the error occurred
615 /// - `module_name`: The name of the module that failed
616 /// - `err`: The error that occurred
617 fn log_module_error(&self, phase: &str, module_name: &str, err: &crate::core::RiError) {
618 let logger = self.ctx.logger();
619 let message = format!("module={module_name} phase={phase} error={err}");
620 let _ = logger.error("Ri.Runtime", message);
621 }
622}
623
624#[cfg(feature = "pyo3")]
625#[pyo3::prelude::pymethods]
626impl RiAppRuntime {
627 fn py_run(&self, callback: Py<pyo3::PyAny>) -> PyResult<()> {
628 let rt = tokio::runtime::Builder::new_current_thread()
629 .enable_all()
630 .build()
631 .map_err::<pyo3::PyErr, _>(|e| e.into())?;
632
633 let runtime = self.clone();
634 pyo3::Python::attach(|py| {
635 rt.block_on(async move {
636 let _ = callback.call0(py);
637 runtime.run(|_ctx| async move { Ok(()) }).await
638 }).map_err(|e| e.into())
639 })
640 }
641
642 fn get_context(&self) -> PyResult<RiServiceContext> {
643 Ok(self.ctx.clone())
644 }
645
646 #[pyo3(name = "logger")]
647 fn logger_py(&self) -> crate::log::RiLogger {
648 self.ctx.logger().clone()
649 }
650}
651
652#[cfg(test)]
653mod tests {
654 use super::*;
655
656 #[test]
657 fn test_app_runtime_creation() {
658 let ctx = RiServiceContext::new_default().expect("default context");
659 let runtime = RiAppRuntime::new(ctx, vec![]);
660 assert!(runtime.modules.try_read().map(|g| g.is_empty()).unwrap_or(false));
661 }
662
663 #[tokio::test]
664 async fn test_app_runtime_run_with_empty_modules() {
665 let ctx = RiServiceContext::new_default().expect("default context");
666 let runtime = RiAppRuntime::new(ctx, vec![]);
667 let result = runtime.run(|_ctx| async { Ok(()) }).await;
668 assert!(result.is_ok());
669 }
670
671 #[test]
672 fn test_app_runtime_clone() {
673 let ctx = RiServiceContext::new_default().expect("default context");
674 let runtime1 = RiAppRuntime::new(ctx, vec![]);
675 let runtime2 = runtime1.clone();
676 assert!(Arc::ptr_eq(&runtime1.modules, &runtime2.modules));
677 }
678}