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

aster_forge_runtime/
startup.rs
