aster_forge_runtime/
startup.rs

1//! Startup phase coordination primitives.
2//!
3//! This module provides the product-neutral mechanics for ordered startup phase execution:
4//! duration collection, optional phase failure handling, report aggregation, and shared tracing.
5//! Product crates still own concrete initialization work such as migrations, cache creation,
6//! driver loading, runtime config reload, audit setup, and application state construction.
7
8use std::future::Future;
9use std::path::PathBuf;
10use std::pin::Pin;
11use std::time::{Duration, Instant};
12
13type StartupFuture = Pin<Box<dyn Future<Output = Result<(), String>> + Send>>;
14type StartupPhaseFn = dyn FnMut() -> StartupFuture + Send;
15
16/// Error returned by runtime temporary directory helpers.
17#[derive(Debug, thiserror::Error)]
18pub enum RuntimeTempDirError {
19    /// The scope segment is empty or contains path-unsafe characters.
20    #[error(
21        "invalid runtime temp scope '{scope}': use non-empty ASCII letters, digits, '-' or '_'"
22    )]
23    InvalidScope {
24        /// Invalid scope value.
25        scope: String,
26    },
27    /// The filesystem operation failed.
28    #[error("runtime temp directory IO failed: {0}")]
29    Io(#[from] std::io::Error),
30}
31
32/// Ensures the short-lived runtime temporary directory exists.
33///
34/// The directory is derived from [`aster_forge_utils::paths::runtime_temp_dir`], so all Aster
35/// services use the same `_runtime` namespace under their configured temporary root. Products keep
36/// ownership of when the directory is cleaned and how IO errors are mapped into their own error
37/// types.
38pub async fn ensure_runtime_temp_dir(temp_root: &str) -> std::io::Result<String> {
39    let runtime_temp_dir = aster_forge_utils::paths::runtime_temp_dir(temp_root);
40    tokio::fs::create_dir_all(&runtime_temp_dir).await?;
41    Ok(runtime_temp_dir)
42}
43
44/// Creates a scope-local runtime temporary directory guarded by RAII cleanup.
45///
46/// The returned [`aster_forge_utils::raii::TempDirGuard`] removes the created directory on drop.
47/// This helper is intended for one operation, such as image rendering, archive extraction, or
48/// temporary external command output. It should not guard the shared `_runtime` root itself.
49pub async fn create_runtime_temp_dir_guard(
50    temp_root: &str,
51    scope: &str,
52    cleanup_label: &'static str,
53) -> Result<aster_forge_utils::raii::TempDirGuard, RuntimeTempDirError> {
54    validate_runtime_temp_scope(scope)?;
55    let runtime_temp_dir = ensure_runtime_temp_dir(temp_root).await?;
56    let scoped_root = aster_forge_utils::paths::join_path(&runtime_temp_dir, scope);
57    let temp_dir = aster_forge_utils::paths::join_path(
58        &scoped_root,
59        &aster_forge_utils::id::new_short_token(),
60    );
61    tokio::fs::create_dir_all(&temp_dir).await?;
62    Ok(aster_forge_utils::raii::TempDirGuard::new(
63        PathBuf::from(temp_dir),
64        cleanup_label,
65    ))
66}
67
68fn validate_runtime_temp_scope(scope: &str) -> Result<(), RuntimeTempDirError> {
69    if scope.is_empty()
70        || !scope
71            .chars()
72            .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_')
73    {
74        return Err(RuntimeTempDirError::InvalidScope {
75            scope: scope.to_string(),
76        });
77    }
78
79    Ok(())
80}
81
82/// Failure policy for one startup phase.
83#[derive(Debug, Clone, Copy, PartialEq, Eq)]
84pub enum StartupPhaseFailurePolicy {
85    /// A failure aborts startup and stops later phases.
86    Required,
87    /// A failure is recorded and startup continues.
88    Optional,
89}
90
91/// Final status for one startup phase.
92#[derive(Debug, Clone, PartialEq, Eq)]
93pub enum StartupPhaseStatus {
94    /// The phase completed successfully.
95    Succeeded,
96    /// A required phase failed and startup stopped.
97    Failed(String),
98    /// An optional phase failed and startup continued.
99    SkippedAfterFailure(String),
100}
101
102impl StartupPhaseStatus {
103    /// Returns whether this phase completed successfully.
104    pub const fn is_success(&self) -> bool {
105        matches!(self, Self::Succeeded)
106    }
107
108    /// Returns whether this phase reported an error.
109    pub const fn is_failure(&self) -> bool {
110        !self.is_success()
111    }
112}
113
114/// Report for one executed startup phase.
115#[derive(Debug, Clone, PartialEq, Eq)]
116pub struct StartupPhaseReport {
117    /// Stable phase name.
118    pub name: &'static str,
119    /// Failure policy used by the phase.
120    pub failure_policy: StartupPhaseFailurePolicy,
121    /// Phase result.
122    pub status: StartupPhaseStatus,
123    /// Execution duration.
124    pub duration: Duration,
125}
126
127/// Aggregate report for a startup run.
128#[derive(Debug, Clone, PartialEq, Eq)]
129pub struct StartupReport {
130    /// Phase reports in execution order.
131    pub phases: Vec<StartupPhaseReport>,
132}
133
134/// Value returned by a startup phase together with its execution report.
135#[derive(Debug, Clone, PartialEq, Eq)]
136pub struct StartupPhaseOutcome<T> {
137    /// Value returned by the product startup phase.
138    pub value: T,
139    /// Report for the executed phase.
140    pub report: StartupPhaseReport,
141}
142
143/// Runs one required startup phase that returns a product-owned value.
144///
145/// This helper is useful for startup steps that construct resources such as database handles,
146/// runtime config snapshots, cache backends, driver registries, or application state. The product
147/// error type is preserved while Forge still provides shared tracing and phase reporting.
148pub async fn run_required_startup_phase<F, Fut, T, E>(
149    name: &'static str,
150    phase: F,
151) -> Result<StartupPhaseOutcome<T>, E>
152where
153    F: FnOnce() -> Fut,
154    Fut: Future<Output = Result<T, E>>,
155    E: std::fmt::Display,
156{
157    tracing::info!(phase = name, "starting startup phase");
158    let started_at = Instant::now();
159    match phase().await {
160        Ok(value) => {
161            let duration = started_at.elapsed();
162            tracing::info!(phase = name, ?duration, "startup phase completed");
163            Ok(StartupPhaseOutcome {
164                value,
165                report: StartupPhaseReport {
166                    name,
167                    failure_policy: StartupPhaseFailurePolicy::Required,
168                    status: StartupPhaseStatus::Succeeded,
169                    duration,
170                },
171            })
172        }
173        Err(error) => {
174            let duration = started_at.elapsed();
175            tracing::error!(phase = name, ?duration, %error, "startup phase failed");
176            Err(error)
177        }
178    }
179}
180
181/// Runs one optional startup phase and returns its report.
182///
183/// Optional phase failures are logged and represented as
184/// [`StartupPhaseStatus::SkippedAfterFailure`], but the error does not abort startup.
185pub async fn run_optional_startup_phase<F, Fut, E>(
186    name: &'static str,
187    phase: F,
188) -> StartupPhaseReport
189where
190    F: FnOnce() -> Fut,
191    Fut: Future<Output = Result<(), E>>,
192    E: std::fmt::Display,
193{
194    tracing::info!(phase = name, "starting startup phase");
195    let started_at = Instant::now();
196    let status = match phase().await {
197        Ok(()) => StartupPhaseStatus::Succeeded,
198        Err(error) => StartupPhaseStatus::SkippedAfterFailure(error.to_string()),
199    };
200    let duration = started_at.elapsed();
201    match &status {
202        StartupPhaseStatus::Succeeded => {
203            tracing::info!(phase = name, ?duration, "startup phase completed");
204        }
205        StartupPhaseStatus::SkippedAfterFailure(error) => {
206            tracing::warn!(
207                phase = name,
208                ?duration,
209                %error,
210                "optional startup phase failed; continuing startup"
211            );
212        }
213        StartupPhaseStatus::Failed(error) => {
214            tracing::error!(phase = name, ?duration, %error, "startup phase failed");
215        }
216    }
217
218    StartupPhaseReport {
219        name,
220        failure_policy: StartupPhaseFailurePolicy::Optional,
221        status,
222        duration,
223    }
224}
225
226impl StartupReport {
227    /// Returns a report from phase entries.
228    pub fn new(phases: Vec<StartupPhaseReport>) -> Self {
229        Self { phases }
230    }
231
232    /// Returns whether startup was aborted by a required phase failure.
233    pub fn aborted(&self) -> bool {
234        self.phases.iter().any(|phase| {
235            matches!(
236                phase.status,
237                StartupPhaseStatus::Failed(_) if phase.failure_policy == StartupPhaseFailurePolicy::Required
238            )
239        })
240    }
241
242    /// Returns whether any phase reported an error.
243    pub fn has_failures(&self) -> bool {
244        self.phases.iter().any(|phase| phase.status.is_failure())
245    }
246}
247
248struct RegisteredStartupPhase {
249    name: &'static str,
250    failure_policy: StartupPhaseFailurePolicy,
251    phase: Box<StartupPhaseFn>,
252}
253
254/// Ordered startup phase coordinator.
255///
256/// The coordinator owns phase ordering, failure policy handling, duration collection, and tracing.
257/// Product crates provide closures for the actual startup work and decide how to map report data
258/// into their own diagnostics or admin surfaces.
259#[derive(Default)]
260pub struct StartupCoordinator {
261    phases: Vec<RegisteredStartupPhase>,
262}
263
264impl StartupCoordinator {
265    /// Creates an empty startup coordinator.
266    pub fn new() -> Self {
267        Self::default()
268    }
269
270    /// Registers a required startup phase.
271    pub fn required<F, Fut>(&mut self, name: &'static str, phase: F) -> &mut Self
272    where
273        F: FnMut() -> Fut + Send + 'static,
274        Fut: Future<Output = Result<(), String>> + Send + 'static,
275    {
276        self.phase(name, StartupPhaseFailurePolicy::Required, phase)
277    }
278
279    /// Registers an optional startup phase.
280    pub fn optional<F, Fut>(&mut self, name: &'static str, phase: F) -> &mut Self
281    where
282        F: FnMut() -> Fut + Send + 'static,
283        Fut: Future<Output = Result<(), String>> + Send + 'static,
284    {
285        self.phase(name, StartupPhaseFailurePolicy::Optional, phase)
286    }
287
288    /// Registers a startup phase with an explicit failure policy.
289    pub fn phase<F, Fut>(
290        &mut self,
291        name: &'static str,
292        failure_policy: StartupPhaseFailurePolicy,
293        mut phase: F,
294    ) -> &mut Self
295    where
296        F: FnMut() -> Fut + Send + 'static,
297        Fut: Future<Output = Result<(), String>> + Send + 'static,
298    {
299        self.phases.push(RegisteredStartupPhase {
300            name,
301            failure_policy,
302            phase: Box::new(move || Box::pin(phase())),
303        });
304        self
305    }
306
307    /// Runs registered phases in order.
308    ///
309    /// Required phase failures stop later phases. Optional phase failures are logged and included in
310    /// the report while later phases continue.
311    pub async fn run(&mut self) -> StartupReport {
312        let mut reports = Vec::with_capacity(self.phases.len());
313
314        for phase in &mut self.phases {
315            tracing::info!(phase = phase.name, "starting startup phase");
316            let started_at = Instant::now();
317            let result = (phase.phase)().await;
318            let duration = started_at.elapsed();
319            let status = match result {
320                Ok(()) => StartupPhaseStatus::Succeeded,
321                Err(error) if phase.failure_policy == StartupPhaseFailurePolicy::Optional => {
322                    StartupPhaseStatus::SkippedAfterFailure(error)
323                }
324                Err(error) => StartupPhaseStatus::Failed(error),
325            };
326
327            match &status {
328                StartupPhaseStatus::Succeeded => {
329                    tracing::info!(phase = phase.name, ?duration, "startup phase completed");
330                }
331                StartupPhaseStatus::SkippedAfterFailure(error) => {
332                    tracing::warn!(
333                        phase = phase.name,
334                        ?duration,
335                        %error,
336                        "optional startup phase failed; continuing startup"
337                    );
338                }
339                StartupPhaseStatus::Failed(error) => {
340                    tracing::error!(phase = phase.name, ?duration, %error, "startup phase failed");
341                }
342            }
343
344            let should_abort = matches!(status, StartupPhaseStatus::Failed(_));
345            reports.push(StartupPhaseReport {
346                name: phase.name,
347                failure_policy: phase.failure_policy,
348                status,
349                duration,
350            });
351            if should_abort {
352                break;
353            }
354        }
355
356        StartupReport::new(reports)
357    }
358
359    /// Returns how many phases are registered.
360    pub fn len(&self) -> usize {
361        self.phases.len()
362    }
363
364    /// Returns whether no phases are registered.
365    pub fn is_empty(&self) -> bool {
366        self.phases.is_empty()
367    }
368}
369
370#[cfg(test)]
371mod tests {
372    use std::path::Path;
373    use std::sync::atomic::{AtomicU64, Ordering};
374
375    use super::{
376        RuntimeTempDirError, StartupCoordinator, StartupPhaseFailurePolicy, StartupPhaseStatus,
377        create_runtime_temp_dir_guard, ensure_runtime_temp_dir, run_optional_startup_phase,
378        run_required_startup_phase,
379    };
380
381    static TEMP_ID: AtomicU64 = AtomicU64::new(0);
382
383    #[tokio::test]
384    async fn ensure_runtime_temp_dir_creates_runtime_namespace() {
385        let root = std::env::temp_dir().join(format!(
386            "aster-forge-runtime-dirs-{}-{}",
387            std::process::id(),
388            TEMP_ID.fetch_add(1, Ordering::Relaxed)
389        ));
390        let root = root.to_string_lossy().to_string();
391
392        let runtime_dir = ensure_runtime_temp_dir(&root)
393            .await
394            .expect("runtime temp dir should be created");
395
396        assert_eq!(
397            runtime_dir,
398            aster_forge_utils::paths::runtime_temp_dir(&root)
399        );
400        assert!(Path::new(&runtime_dir).is_dir());
401        let _ = std::fs::remove_dir_all(root);
402    }
403
404    #[tokio::test]
405    async fn create_runtime_temp_dir_guard_creates_scope_local_directory_and_cleans_on_drop() {
406        let root = std::env::temp_dir().join(format!(
407            "aster-forge-runtime-guard-{}-{}",
408            std::process::id(),
409            TEMP_ID.fetch_add(1, Ordering::Relaxed)
410        ));
411        let root = root.to_string_lossy().to_string();
412        let guarded_path;
413
414        {
415            let guard = create_runtime_temp_dir_guard(&root, "thumbnail", "test runtime temp dir")
416                .await
417                .expect("runtime temp dir guard should be created");
418            guarded_path = guard.path().to_path_buf();
419
420            assert!(guarded_path.is_dir());
421            assert!(guarded_path.starts_with(aster_forge_utils::paths::runtime_temp_dir(&root)));
422            assert!(guarded_path.parent().is_some_and(|parent| {
423                parent.ends_with(aster_forge_utils::paths::join_path(
424                    &aster_forge_utils::paths::runtime_temp_dir(&root),
425                    "thumbnail",
426                ))
427            }));
428        }
429
430        assert!(!guarded_path.exists());
431        let _ = std::fs::remove_dir_all(root);
432    }
433
434    #[tokio::test]
435    async fn create_runtime_temp_dir_guard_rejects_path_like_scope() {
436        let error = match create_runtime_temp_dir_guard("target/tmp", "../bad", "test").await {
437            Ok(_) => {
438                panic!("path-like scope should be rejected");
439            }
440            Err(error) => error,
441        };
442
443        assert!(matches!(error, RuntimeTempDirError::InvalidScope { .. }));
444    }
445
446    #[tokio::test]
447    async fn startup_coordinator_runs_required_phases_in_order() {
448        let mut coordinator = StartupCoordinator::new();
449        coordinator
450            .required("database", || async { Ok(()) })
451            .required("cache", || async { Ok(()) });
452
453        let report = coordinator.run().await;
454
455        assert_eq!(coordinator.len(), 2);
456        assert!(!report.has_failures());
457        assert!(!report.aborted());
458        assert_eq!(report.phases[0].name, "database");
459        assert_eq!(report.phases[1].name, "cache");
460    }
461
462    #[tokio::test]
463    async fn startup_coordinator_aborts_after_required_failure() {
464        let mut coordinator = StartupCoordinator::new();
465        coordinator
466            .required("database", || async {
467                Err("database unavailable".to_string())
468            })
469            .required("cache", || async { Ok(()) });
470
471        let report = coordinator.run().await;
472
473        assert!(report.has_failures());
474        assert!(report.aborted());
475        assert_eq!(report.phases.len(), 1);
476        assert_eq!(
477            report.phases[0].status,
478            StartupPhaseStatus::Failed("database unavailable".to_string())
479        );
480    }
481
482    #[tokio::test]
483    async fn startup_coordinator_continues_after_optional_failure() {
484        let mut coordinator = StartupCoordinator::new();
485        coordinator
486            .optional("metrics", || async {
487                Err("prometheus unavailable".to_string())
488            })
489            .required("database", || async { Ok(()) });
490
491        let report = coordinator.run().await;
492
493        assert!(report.has_failures());
494        assert!(!report.aborted());
495        assert_eq!(report.phases.len(), 2);
496        assert_eq!(
497            report.phases[0].status,
498            StartupPhaseStatus::SkippedAfterFailure("prometheus unavailable".to_string())
499        );
500        assert_eq!(report.phases[1].status, StartupPhaseStatus::Succeeded);
501    }
502
503    #[tokio::test]
504    async fn run_required_startup_phase_returns_product_value() {
505        let outcome =
506            run_required_startup_phase("build_state", || async { Ok::<_, String>("state") })
507                .await
508                .expect("phase should succeed");
509
510        assert_eq!(outcome.value, "state");
511        assert_eq!(outcome.report.name, "build_state");
512        assert_eq!(outcome.report.status, StartupPhaseStatus::Succeeded);
513        assert_eq!(
514            outcome.report.failure_policy,
515            StartupPhaseFailurePolicy::Required
516        );
517    }
518
519    #[tokio::test]
520    async fn run_required_startup_phase_preserves_product_error() {
521        let error = run_required_startup_phase("build_state", || async {
522            Err::<(), _>("database unavailable")
523        })
524        .await
525        .expect_err("phase should return product error");
526
527        assert_eq!(error, "database unavailable");
528    }
529
530    #[tokio::test]
531    async fn run_optional_startup_phase_reports_failure_without_returning_error() {
532        let report = run_optional_startup_phase("metrics", || async {
533            Err::<(), _>("prometheus unavailable")
534        })
535        .await;
536
537        assert_eq!(report.name, "metrics");
538        assert_eq!(report.failure_policy, StartupPhaseFailurePolicy::Optional);
539        assert_eq!(
540            report.status,
541            StartupPhaseStatus::SkippedAfterFailure("prometheus unavailable".to_string())
542        );
543    }
544}
aster_forge_runtime/startup.rs

aster_forge_runtime/
startup.rs
