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

aster_forge_runtime/
startup.rs
