saddle-framework 0.3.21

//! Facade-owned startup staging. Runtime owns the eventual close deadline.
use super::*;
use saddle_core::{
    CaptureSite, Diagnostic, DiagnosticCategory, DiagnosticCause, DiagnosticCode, DiagnosticObject,
    DiagnosticObjectKind, DiagnosticStage,
};
use saddle_observability::{
    DiagnosticSubmission, EmergencyDiagnostics, EmergencyInitError, FileLoggingConfig, Rotation,
};

static OUTPUT: std::sync::RwLock<Option<saddle_observability::EmergencyDiagnosticHandle>> =
    std::sync::RwLock::new(None);

pub(super) type SharedOutput = Arc<Mutex<ProductionOutput>>;
pub(super) struct ProductionOutput {
    owner: Option<EmergencyDiagnostics>,
    _registration: Option<OutputRegistration>,
    exit: Option<saddle_runtime::diagnostics::RuntimeDiagnosticExit>,
}

pub(super) fn load_production<B: DeserializeOwned + 'static>(
    path: &Path,
) -> Result<ProcessConfig<B>> {
    let prepared = PreparedStartup::<B>::load(path).map_err(|error| {
        // The logging subtree itself could not be resolved. No other directory
        // can safely be invented; this is the explicit last-resort exit.
        eprintln!("{error}; diagnostic_output=unavailable_config");
        error
    })?;
    let PreparedStartup {
        output,
        config,
        registration,
        ..
    } = prepared;
    let owner = match output {
        Ok(owner) => owner,
        Err(output_error) => {
            return match config {
                Err(primary) => {
                    eprintln!("{primary}; output_failure={output_error}");
                    Err(primary)
                }
                Ok(_) => {
                    eprintln!("{output_error}");
                    Err(output_error)
                }
            };
        }
    };
    let shared = Arc::new(Mutex::new(ProductionOutput {
        owner: Some(owner),
        _registration: registration,
        exit: None,
    }));
    let mut config = match config {
        Ok(config) => config,
        Err(error) => return finish(Err(error), Some(&shared)),
    };
    let handle = shared.lock().unwrap().owner.as_ref().unwrap().handle();
    if saddle_runtime::diagnostics::install_output(handle).is_err() {
        return finish(
            Err(report(config_failure(
                "saddle.process.diagnostic_output_already_installed",
                None,
            ))),
            Some(&shared),
        );
    }
    static HOOK: std::sync::Once = std::sync::Once::new();
    HOOK.call_once(|| {
        std::panic::set_hook(Box::new(|info| {
            if !saddle_runtime::diagnostics::capture_current_panic(info) {
                let diagnostic = Diagnostic::capture_panic(info, DiagnosticStage::BackgroundTask);
                let _submission = record(&diagnostic, None);
            }
            // Never chain the default hook: it prints the arbitrary panic payload.
        }))
    });
    config.diagnostics = Some(shared);
    Ok(config)
}

pub(super) fn take_owner(output: &SharedOutput) -> Option<EmergencyDiagnostics> {
    output
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .owner
        .take()
}

pub(super) fn parse_error(path: &Path, source: &str, error: &toml::de::Error) -> SaddleError {
    let (line, column) = error
        .span()
        .map(|span| {
            let prefix = &source.as_bytes()[..span.start.min(source.len())];
            let line = prefix.iter().filter(|&&b| b == b'\n').count() as u64 + 1;
            let column = prefix
                .iter()
                .rposition(|&b| b == b'\n')
                .map_or(prefix.len(), |last| prefix.len() - last - 1)
                as u64
                + 1;
            (Some(line), Some(column))
        })
        .unwrap_or((None, None));
    let file = path
        .file_name()
        .and_then(|p| p.to_str())
        .and_then(|name| saddle_core::DiagnosticLocator::from_projection(name, false, false))
        .unwrap_or_else(|| {
            saddle_core::DiagnosticLocator::from_projection("", false, true)
                .expect("redacted locator")
        });
    let location = saddle_core::DiagnosticInputLocation::new(line, column).with_file(file);
    let cause = DiagnosticCause::new(
        DiagnosticStage::StartupConfig,
        DiagnosticCode::new("saddle.process.config_invalid").unwrap(),
    )
    .with_input_location(location);
    // TOML's formatted message may contain configuration values; retain its
    // input span, never its raw Display/Debug. Missing span stays explicitly None.
    SaddleError::new(
        ErrorKind::InvalidArgument,
        "saddle.process.config_invalid",
        "configuration parse failed",
    )
    .with_diagnostic(Diagnostic::capture(
        DiagnosticCategory::UnexpectedError,
        CaptureSite::FirstObserved,
        cause,
    ))
}
pub(super) fn set_exit(
    output: &SharedOutput,
    exit: saddle_runtime::diagnostics::RuntimeDiagnosticExit,
) {
    output.lock().unwrap_or_else(|e| e.into_inner()).exit = Some(exit);
}

#[track_caller]
pub(super) fn report(error: SaddleError) -> SaddleError {
    let error = if error.diagnostic().is_some() {
        error
    } else {
        let cause = DiagnosticCause::new(
            DiagnosticStage::StartupConfig,
            DiagnosticCode::new(error.code()).unwrap_or_else(|| {
                DiagnosticCode::new("saddle.process.unclassified_failure").unwrap()
            }),
        );
        error.with_diagnostic(Diagnostic::capture(
            DiagnosticCategory::UnexpectedError,
            CaptureSite::FirstObserved,
            cause,
        ))
    };
    if let Some(diagnostic) = error.diagnostic() {
        let _submission = record(diagnostic, None);
    }
    error
}

pub(super) fn finish<T>(result: Result<T>, output: Option<&SharedOutput>) -> Result<T> {
    if let Some(output) = output {
        let mut output = output.lock().unwrap_or_else(|e| e.into_inner());
        if let Some(mut owner) = output.owner.take() {
            // No Application shutdown deadline exists on a pre-coordinator
            // rejection. Observe once, exactly as Runtime's no-deadline path.
            let shutdown = owner.shutdown();
            output.exit = Some(saddle_runtime::diagnostics::RuntimeDiagnosticExit {
                shutdown,
                snapshot: owner.snapshot(),
            });
        }
        if let Some(exit) = &output.exit {
            let state = exit.snapshot;
            if !state.initialized
                || state.first_failure.is_some()
                || state.written != state.enqueued
                || state.dropped != 0
                || exit.shutdown != saddle_observability::DiagnosticShutdown::Finished
            {
                // Safe counters/classification only. Never replace primary error
                // with logging failure or claim Pending means written.
                eprintln!(
                    "saddle.diagnostic.output_not_confirmed shutdown={:?} initialized={} enqueued={} written={} dropped={} first_failure={:?}",
                    exit.shutdown,
                    state.initialized,
                    state.enqueued,
                    state.written,
                    state.dropped,
                    state.first_failure
                );
                if let Err(error) = &result {
                    eprintln!("{error}");
                }
            }
        }
    }
    result
}

pub(crate) struct OutputRegistration;
impl OutputRegistration {
    pub(crate) fn install(output: &EmergencyDiagnostics) -> Self {
        // EmergencyDiagnostics already enforces one active process writer.
        *OUTPUT.write().unwrap_or_else(|e| e.into_inner()) = Some(output.handle());
        Self
    }
}
impl Drop for OutputRegistration {
    fn drop(&mut self) {
        *OUTPUT.write().unwrap_or_else(|e| e.into_inner()) = None;
    }
}

/// Same process output, with only the source-provided checked association.
/// None means no active Facade-owned output; it is not delivery success.
pub(crate) fn record(
    diagnostic: &Diagnostic,
    context: Option<(
        &saddle_core::CallContext,
        &saddle_observability::EventContext,
    )>,
) -> Option<DiagnosticSubmission> {
    let handle = OUTPUT.read().unwrap_or_else(|e| e.into_inner()).clone()?;
    Some(match saddle_observability::global() {
        Some(observer) => observer.record_diagnostic(diagnostic, &handle, context),
        None => handle.submit_context(diagnostic, context),
    })
}

// Only the existing logging subtree is selected here. The complete, strict
// configuration parser still validates the SAME source below. No fallback to a
// different directory when this subtree is invalid.
#[derive(Deserialize)]
struct EarlyFile {
    framework: EarlyFramework,
}
#[derive(Deserialize)]
struct EarlyFramework {
    #[serde(default)]
    observability: ObservabilityFileConfig,
}

pub(super) struct PreparedStartup<B> {
    pub output: Result<EmergencyDiagnostics>,
    pub config: Result<ProcessConfig<B>>,
    pub submission: Option<DiagnosticSubmission>,
    pub registration: Option<OutputRegistration>,
}

impl<B: DeserializeOwned + 'static> PreparedStartup<B> {
    pub fn load(path: &Path) -> Result<Self> {
        let bytes = std::fs::read(path)
            .map_err(|error| config_failure("saddle.process.config_unavailable", Some(&error)))?;
        let source = std::str::from_utf8(&bytes)
            .map_err(|_| config_failure("saddle.process.config_invalid_utf8", None))?;
        let early: EarlyFile = toml::from_str(source)
            .map_err(|_| config_failure("saddle.process.logging_configuration_unresolved", None))?;
        let logging = early.framework.observability.logging;
        let logging = FileLoggingConfig::new(
            logging.directory.unwrap_or_else(|| PathBuf::from("./logs")),
            match logging.rotation {
                LoggingRotation::Daily => Rotation::Daily,
                LoggingRotation::Hourly => Rotation::Hourly,
            },
        );
        let output = EmergencyDiagnostics::start(&logging).map_err(|error| {
            let code = match error {
                EmergencyInitError::InvalidDirectory => {
                    "saddle.process.diagnostic_directory_invalid"
                }
                EmergencyInitError::AlreadyActive => {
                    "saddle.process.diagnostic_writer_already_active"
                }
                EmergencyInitError::Spawn(_) => "saddle.process.diagnostic_worker_spawn_failed",
            };
            config_failure(code, None)
        });
        let registration = output.as_ref().ok().map(OutputRegistration::install);
        let config = ProcessConfig::load_source(path, source).map_err(|error| {
            if error.diagnostic().is_some() {
                error
            } else {
                let cause = DiagnosticCause::new(
                    DiagnosticStage::StartupConfig,
                    DiagnosticCode::new(error.code()).expect("framework static code"),
                );
                // No arbitrary parser Display/Debug or config values. Precise
                // input location is integrated with the DB/Obs follow-up.
                error.with_diagnostic(Diagnostic::capture(
                    DiagnosticCategory::UnexpectedError,
                    CaptureSite::FirstObserved,
                    cause,
                ))
            }
        });
        let submission = config
            .as_ref()
            .err()
            .and_then(SaddleError::diagnostic)
            .and_then(|d| output.as_ref().ok().map(|output| output.handle().submit(d)));
        Ok(Self {
            output,
            config,
            submission,
            registration,
        })
    }
}

#[track_caller]
fn config_failure(code: &'static str, io: Option<&std::io::Error>) -> SaddleError {
    let mut cause = DiagnosticCause::new(
        DiagnosticStage::StartupConfig,
        DiagnosticCode::new(code).expect("static code"),
    )
    .with_object(
        DiagnosticObject::new(
            DiagnosticObjectKind::ConfigKey,
            "framework.observability.logging",
        )
        .expect("static key"),
    );
    if let Some(io) = io {
        cause = cause.with_io(io);
    }
    SaddleError::new(
        ErrorKind::Infrastructure,
        code,
        "startup configuration could not establish diagnostic output",
    )
    .with_diagnostic(Diagnostic::capture(
        DiagnosticCategory::UnexpectedError,
        CaptureSite::FirstObserved,
        cause,
    ))
}

// This staging owner intentionally has no blocking Drop or extra timeout.
// Generated entry integration must close it under Runtime's original budget.

#[cfg(test)]
mod tests {
    use super::*;
    use saddle_observability::DiagnosticShutdown;

    #[test]
    fn early_config_failure_uses_selected_directory_and_preserves_primary() {
        let root = std::env::temp_dir().join(format!("saddle-early-config-{}", std::process::id()));
        std::fs::create_dir(&root).unwrap();
        let path = root.join("saddle.toml");
        let base = "[framework]\nlisten='127.0.0.1:0'\n[framework.management]\nbind='127.0.0.1:0'\n[framework.admission]\ncpuCores=2\nmemoryMb=512\n[framework.admission.dependencies]\ndatabaseConcurrency=1\nprofusecontractConcurrency=1\n[framework.profusecontract]\nauthority='http://localhost:50051'\ntoken='DO_NOT_LOG_TOKEN_68142'\n[secrets]\n";
        for (name, bad_config, bad_directory) in [
            ("good logs", false, false),
            ("invalid config logs", true, false),
            ("file-not-directory", true, true),
        ] {
            let directory = root.join(name);
            if bad_directory {
                std::fs::write(&directory, "not a directory").unwrap();
            }
            let source = format!(
                "{base}[framework.observability.logging]\ndirectory={}\n{}",
                serde_json::to_string(&directory).unwrap(),
                if bad_config {
                    "[business]\ninvalid='DO_NOT_LOG_DATA_93715'\n"
                } else {
                    ""
                }
            );
            std::fs::write(&path, source).unwrap();
            let mut prepared = PreparedStartup::<()>::load(&path).unwrap();
            assert_eq!(prepared.config.is_err(), bad_config);
            if bad_config {
                assert_eq!(
                    prepared.config.as_ref().err().unwrap().code(),
                    "saddle.process.config_invalid"
                );
            }
            let output = prepared.output.as_mut().unwrap();
            let deadline = std::time::Instant::now() + Duration::from_secs(5);
            while !output.snapshot().initialized && std::time::Instant::now() < deadline {
                std::thread::sleep(Duration::from_millis(1));
            }
            assert!(output.snapshot().initialized);
            assert_eq!(output.snapshot().first_failure.is_some(), bad_directory);
            while output.shutdown() == DiagnosticShutdown::Pending
                && std::time::Instant::now() < deadline
            {
                std::thread::sleep(Duration::from_millis(1));
            }
            assert_eq!(output.shutdown(), DiagnosticShutdown::Finished);
            if bad_config && !bad_directory {
                assert_eq!(prepared.submission, Some(DiagnosticSubmission::Enqueued));
                assert_eq!(output.snapshot().written, 1);
                let record = std::fs::read_to_string(output.target()).unwrap();
                assert!(record.contains("saddle.process.config_invalid"));
                assert!(!record.contains("DO_NOT_LOG_TOKEN_68142"));
                assert!(!record.contains("DO_NOT_LOG_DATA_93715"));
            }
            if !bad_directory {
                std::fs::remove_file(output.target()).unwrap();
                std::fs::remove_dir(directory).unwrap();
            } else {
                assert!(prepared.config.is_err());
                std::fs::remove_file(directory).unwrap();
            }
        }
        std::fs::remove_file(path).unwrap();
        std::fs::remove_dir(root).unwrap();
    }
}