diff --git a/crates/lantern-app/src/backup_flow.rs b/crates/lantern-app/src/backup_flow.rs new file mode 100644 index 00000000..6f14f798 --- /dev/null +++ b/crates/lantern-app/src/backup_flow.rs @@ -0,0 +1,175 @@ +use std::path::PathBuf; + +use lantern_domain::{BackupDiffStatus, BackupDifference, BackupSnapshot, DeviceWriteOutcome}; + +use crate::{ApprovedRestorePlan, BackupCaptureContext, RestoreConfirmation}; + +#[derive(Clone, Debug)] +pub struct StoredBackup { + pub path: PathBuf, + pub snapshot: BackupSnapshot, +} + +#[derive(Clone, Debug)] +pub struct PreparedRestoreBundle { + pub pre_restore: StoredBackup, + pub diff: Vec, + pub plan: ApprovedRestorePlan, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct RestoreExecutionSummary { + pub attempted_steps: usize, + pub verified_steps: usize, + pub terminal_outcome: Option, +} + +#[derive(Clone, Debug)] +pub enum BackupAction { + RefreshCatalog, + CatalogRefreshed(Result, String>), + Capture, + Captured(Result), + SelectSource(PathBuf), + SourceLoaded { + path: PathBuf, + result: Result, + }, + ClearSource, + PrepareRestore, + RestorePrepared(Box>), + ConfirmRestore { + operator_text: String, + }, + RestoreCompleted(Result), +} + +#[derive(Clone, Debug)] +pub enum BackupEffect { + RefreshCatalog, + Capture { + context: BackupCaptureContext, + }, + LoadSource { + path: PathBuf, + }, + PrepareRestore { + source: Box, + context: BackupCaptureContext, + }, + ExecuteRestore { + plan: ApprovedRestorePlan, + confirmation: RestoreConfirmation, + }, +} + +#[derive(Clone, Debug, Default)] +pub(crate) struct BackupRestoreState { + pub catalog: Vec, + pub last_capture: Option, + pub source: Option, + pub pre_restore: Option, + pub diff: Vec, + pub prepared_plan: Option, + pub status: Option, + pub error: Option, +} + +impl BackupRestoreState { + pub(crate) fn invalidate_prepared_operation(&mut self) { + self.pre_restore = None; + self.diff.clear(); + self.prepared_plan = None; + } + + pub(crate) fn view(&self) -> BackupRestoreView { + BackupRestoreView { + catalog: self.catalog.clone(), + last_capture: self.last_capture.as_ref().map(backup_summary), + source: self.source.as_ref().map(backup_summary), + pre_restore: self.pre_restore.as_ref().map(backup_summary), + diff: self + .diff + .iter() + .map(|entry| BackupDiffEntryView { + parameter_id: entry.parameter_id.as_str().to_owned(), + status: entry.status, + }) + .collect(), + prepared_plan: self.prepared_plan.as_ref().map(|plan| RestorePlanView { + plan_hash: plan.plan_hash().to_owned(), + challenge: plan.operator_confirmation_text(), + steps: plan + .steps() + .iter() + .map(|step| RestoreStepView { + index: step.index(), + parameter_id: step.parameter_id().as_str().to_owned(), + expected_old: format!("{:?}", step.expected_old_raw().as_slice()), + target: format!("{:?}", step.target_raw().as_slice()), + }) + .collect(), + skipped: plan.skipped().len(), + }), + status: self.status.clone(), + error: self.error.clone(), + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct BackupSummaryView { + pub path: PathBuf, + pub backup_id: u128, + pub complete: bool, + pub profile_id: String, + pub profile_hash: String, + pub values: usize, + pub errors: usize, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct BackupDiffEntryView { + pub parameter_id: String, + pub status: BackupDiffStatus, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct RestoreStepView { + pub index: usize, + pub parameter_id: String, + pub expected_old: String, + pub target: String, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct RestorePlanView { + pub plan_hash: String, + pub challenge: String, + pub steps: Vec, + pub skipped: usize, +} + +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct BackupRestoreView { + pub catalog: Vec, + pub last_capture: Option, + pub source: Option, + pub pre_restore: Option, + pub diff: Vec, + pub prepared_plan: Option, + pub status: Option, + pub error: Option, +} + +fn backup_summary(stored: &StoredBackup) -> BackupSummaryView { + BackupSummaryView { + path: stored.path.clone(), + backup_id: stored.snapshot.backup_id.get(), + complete: stored.snapshot.is_complete(), + profile_id: stored.snapshot.profile_id.as_str().to_owned(), + profile_hash: stored.snapshot.profile_hash.clone(), + values: stored.snapshot.values.len(), + errors: stored.snapshot.errors.len(), + } +} diff --git a/crates/lantern-app/src/lib.rs b/crates/lantern-app/src/lib.rs index fc836310..4bd98990 100644 --- a/crates/lantern-app/src/lib.rs +++ b/crates/lantern-app/src/lib.rs @@ -4,6 +4,8 @@ mod application; mod backup; +#[cfg(not(test))] +mod backup_flow; mod bus; mod clock; mod connection; @@ -17,6 +19,8 @@ mod monitoring_projection; mod parameters; mod poll; mod ports; +#[cfg(not(test))] +mod product_application; mod profile_registry; mod restore; mod restore_permit; @@ -27,8 +31,18 @@ mod telemetry; mod write_coordinator; mod write_flow; +#[cfg(test)] pub use application::*; +#[cfg(not(test))] +#[doc(hidden)] +pub use application::{ + ApplicationRuntime as LegacyApplicationRuntime, EffectRunner as LegacyEffectRunner, +}; +#[cfg(not(test))] +pub use application::{AuditHealthView, AuthorizationView, OperationView}; pub use backup::*; +#[cfg(not(test))] +pub use backup_flow::*; pub use bus::*; pub use clock::*; pub use connection::*; @@ -54,6 +68,8 @@ pub use monitoring_projection::*; pub use parameters::*; pub use poll::*; pub use ports::*; +#[cfg(not(test))] +pub use product_application::*; pub use profile_registry::*; pub use restore::*; pub use restore_permit::*; diff --git a/crates/lantern-app/src/product_application.rs b/crates/lantern-app/src/product_application.rs new file mode 100644 index 00000000..4940426c --- /dev/null +++ b/crates/lantern-app/src/product_application.rs @@ -0,0 +1,483 @@ +use std::{ + path::PathBuf, + sync::Arc, + time::{SystemTime, UNIX_EPOCH}, +}; + +use lantern_domain::{DriveState, ProfileId, SlaveId, UtcTimestamp}; + +use crate::{ + BackupAction, BackupCaptureContext, BackupEffect, BackupRestoreView, ConnectionAction, + ConnectionWizardView, FaultAction, FaultTimelineView, MonitoringAction, MonitoringView, + ParameterAction, ParameterBrowserView, ProfileRegistry, RestoreConfirmation, SessionInput, + SessionStateMachine, + backup_flow::{BackupRestoreState, PreparedRestoreBundle}, +}; + +use crate::application as legacy; + +pub use legacy::{ApplicationEffectError, SessionPhaseView, SessionView}; + +#[derive(Clone, Debug)] +pub enum ApplicationAction { + ReplaceRegistry(Arc), + SelectProfile(ProfileId), + Connection(ConnectionAction), + Monitoring(MonitoringAction), + Parameters(ParameterAction), + Faults(FaultAction), + Session(SessionInput), + Backup(Box), +} + +#[derive(Clone, Debug)] +pub enum ApplicationEffect { + Connection(crate::ConnectionEffect), + Monitoring(crate::MonitoringEffect), + Faults(crate::FaultEffect), + Write(crate::WriteEffect), + Session(crate::SessionEffect), + Backup(Box), +} + +impl From for ApplicationEffect { + fn from(effect: legacy::ApplicationEffect) -> Self { + match effect { + legacy::ApplicationEffect::Connection(effect) => Self::Connection(effect), + legacy::ApplicationEffect::Monitoring(effect) => Self::Monitoring(effect), + legacy::ApplicationEffect::Faults(effect) => Self::Faults(effect), + legacy::ApplicationEffect::Write(effect) => Self::Write(effect), + legacy::ApplicationEffect::Session(effect) => Self::Session(effect), + } + } +} + +pub trait EffectRunner { + fn execute(&mut self, effect: ApplicationEffect) -> Result<(), ApplicationEffectError>; +} + +pub struct ApplicationRuntime { + state: ApplicationState, + runner: R, +} + +impl ApplicationRuntime { + #[must_use] + pub fn new(state: ApplicationState, runner: R) -> Self { + Self { state, runner } + } + + pub fn dispatch(&mut self, action: ApplicationAction) -> Result<(), ApplicationEffectError> { + for effect in self.state.reduce(action) { + self.runner.execute(effect)?; + } + Ok(()) + } + + #[must_use] + pub const fn state(&self) -> &ApplicationState { + &self.state + } +} + +pub struct ApplicationState { + inner: legacy::ApplicationState, + backup: BackupRestoreState, + build_id: String, +} + +impl Default for ApplicationState { + fn default() -> Self { + Self { + inner: legacy::ApplicationState::default(), + backup: BackupRestoreState::default(), + build_id: "development".to_owned(), + } + } +} + +impl ApplicationState { + #[must_use] + pub fn with_registry(registry: Arc, process_writes_enabled: bool) -> Self { + Self { + inner: legacy::ApplicationState::with_registry(registry, process_writes_enabled), + ..Self::default() + } + } + + #[must_use] + pub fn with_registry_and_suggestions( + registry: Arc, + process_writes_enabled: bool, + suggested_device: Option, + suggested_slave: Option, + ) -> Self { + Self { + inner: legacy::ApplicationState::with_registry_and_suggestions( + registry, + process_writes_enabled, + suggested_device, + suggested_slave, + ), + ..Self::default() + } + } + + pub fn set_build_id(&mut self, build_id: impl Into) { + self.build_id = build_id.into(); + } + + #[must_use] + pub fn view(&self) -> ApplicationView { + ApplicationView { + inner: self.inner.view(), + backup: self.backup.view(), + } + } + + #[must_use] + pub fn registry(&self) -> &Arc { + self.inner.registry() + } + + #[must_use] + pub const fn session(&self) -> &SessionStateMachine { + self.inner.session() + } + + pub fn reduce(&mut self, action: ApplicationAction) -> Vec { + if let ApplicationAction::Backup(action) = action { + return self.reduce_backup(*action); + } + + let previous_session = self.inner.view().active_session(); + let legacy_action = match action { + ApplicationAction::ReplaceRegistry(value) => { + legacy::ApplicationAction::ReplaceRegistry(value) + } + ApplicationAction::SelectProfile(value) => { + legacy::ApplicationAction::SelectProfile(value) + } + ApplicationAction::Connection(value) => legacy::ApplicationAction::Connection(value), + ApplicationAction::Monitoring(value) => legacy::ApplicationAction::Monitoring(value), + ApplicationAction::Parameters(value) => legacy::ApplicationAction::Parameters(value), + ApplicationAction::Faults(value) => legacy::ApplicationAction::Faults(value), + ApplicationAction::Session(value) => legacy::ApplicationAction::Session(value), + ApplicationAction::Backup(_) => unreachable!(), + }; + let effects = self + .inner + .reduce(legacy_action) + .into_iter() + .map(ApplicationEffect::from) + .collect::>(); + let current_view = self.inner.view(); + if current_view.active_session() != previous_session + || current_view.session().phase() != SessionPhaseView::Connected + { + self.backup.invalidate_prepared_operation(); + } + effects + } + + fn reduce_backup(&mut self, action: BackupAction) -> Vec { + match action { + BackupAction::RefreshCatalog => { + self.backup.status = Some("refreshing backup catalog".to_owned()); + self.backup.error = None; + vec![ApplicationEffect::Backup(Box::new( + BackupEffect::RefreshCatalog, + ))] + } + BackupAction::CatalogRefreshed(result) => { + match result { + Ok(paths) => { + self.backup.catalog = paths; + self.backup.status = Some(format!( + "backup catalog refreshed: {} file(s)", + self.backup.catalog.len() + )); + self.backup.error = None; + } + Err(error) => { + self.backup.status = None; + self.backup.error = Some(error); + } + } + Vec::new() + } + BackupAction::Capture => match self.backup_capture_context() { + Ok(context) => { + self.backup.status = Some("capturing complete profile backup".to_owned()); + self.backup.error = None; + vec![ApplicationEffect::Backup(Box::new(BackupEffect::Capture { + context, + }))] + } + Err(error) => { + self.backup.status = None; + self.backup.error = Some(error); + Vec::new() + } + }, + BackupAction::Captured(result) => { + match result { + Ok(stored) => { + if !self.backup.catalog.contains(&stored.path) { + self.backup.catalog.push(stored.path.clone()); + self.backup.catalog.sort(); + } + self.backup.last_capture = Some(stored); + self.backup.status = Some("backup capture persisted".to_owned()); + self.backup.error = None; + } + Err(error) => { + self.backup.status = None; + self.backup.error = Some(error); + } + } + Vec::new() + } + BackupAction::SelectSource(path) => { + self.backup.invalidate_prepared_operation(); + self.backup.status = Some(format!("loading backup {}", path.display())); + self.backup.error = None; + vec![ApplicationEffect::Backup(Box::new( + BackupEffect::LoadSource { path }, + ))] + } + BackupAction::SourceLoaded { path, result } => { + match result { + Ok(snapshot) => { + self.backup.source = Some(crate::StoredBackup { path, snapshot }); + self.backup.status = Some("restore source backup loaded".to_owned()); + self.backup.error = None; + } + Err(error) => { + self.backup.source = None; + self.backup.status = None; + self.backup.error = Some(error); + } + } + Vec::new() + } + BackupAction::ClearSource => { + self.backup.source = None; + self.backup.invalidate_prepared_operation(); + self.backup.status = Some("restore source cleared".to_owned()); + self.backup.error = None; + Vec::new() + } + BackupAction::PrepareRestore => { + let Some(source) = self + .backup + .source + .as_ref() + .map(|stored| stored.snapshot.clone()) + else { + self.backup.error = + Some("select a source backup before preparing restore".to_owned()); + return Vec::new(); + }; + match self.backup_capture_context() { + Ok(context) => { + self.backup.invalidate_prepared_operation(); + self.backup.status = Some( + "capturing fresh pre-restore backup and building guarded plan" + .to_owned(), + ); + self.backup.error = None; + vec![ApplicationEffect::Backup(Box::new( + BackupEffect::PrepareRestore { + source: Box::new(source), + context, + }, + ))] + } + Err(error) => { + self.backup.status = None; + self.backup.error = Some(error); + Vec::new() + } + } + } + BackupAction::RestorePrepared(result) => { + match *result { + Ok(PreparedRestoreBundle { + pre_restore, + diff, + plan, + }) => { + let steps = plan.steps().len(); + self.backup.pre_restore = Some(pre_restore); + self.backup.diff = diff; + self.backup.prepared_plan = Some(plan); + self.backup.status = Some(format!( + "guarded restore plan prepared: {steps} step(s); exact confirmation required" + )); + self.backup.error = None; + } + Err(error) => { + self.backup.invalidate_prepared_operation(); + self.backup.status = None; + self.backup.error = Some(error); + } + } + Vec::new() + } + BackupAction::ConfirmRestore { operator_text } => { + let Some(plan) = self.backup.prepared_plan.as_ref() else { + self.backup.error = Some("there is no prepared restore plan".to_owned()); + return Vec::new(); + }; + if operator_text != plan.operator_confirmation_text() { + self.backup.error = Some( + "operator confirmation does not exactly match the restore plan".to_owned(), + ); + return Vec::new(); + } + let plan = self + .backup + .prepared_plan + .take() + .expect("prepared plan checked above"); + self.backup.status = Some("executing guarded restore".to_owned()); + self.backup.error = None; + vec![ApplicationEffect::Backup(Box::new( + BackupEffect::ExecuteRestore { + confirmation: RestoreConfirmation::Confirm { + challenge: operator_text, + }, + plan, + }, + ))] + } + BackupAction::RestoreCompleted(result) => { + self.backup.prepared_plan = None; + match result { + Ok(summary) => { + self.backup.status = Some(format!( + "restore finished: attempted={} verified={} terminal={:?}", + summary.attempted_steps, + summary.verified_steps, + summary.terminal_outcome + )); + self.backup.error = None; + } + Err(error) => { + self.backup.status = None; + self.backup.error = Some(error); + } + } + Vec::new() + } + } + } + + fn backup_capture_context(&self) -> Result { + let view = self.inner.view(); + if view.session().phase() != SessionPhaseView::Connected || view.active_session().is_none() + { + return Err("backup/restore requires a connected Verified session".to_owned()); + } + let profile_hash = view + .session() + .profile_hash() + .ok_or_else(|| "Verified session has no profile hash".to_owned())?; + let entry = self + .inner + .registry() + .find_by_hash(profile_hash) + .ok_or_else(|| "active validated profile is unavailable".to_owned())?; + let link = view + .connection() + .link + .as_ref() + .ok_or_else(|| "active connection has no validated link settings".to_owned())?; + let now = utc_now(); + Ok(BackupCaptureContext { + app_version: env!("CARGO_PKG_VERSION").to_owned(), + build_id: self.build_id.clone(), + profile_origin: format!("{:?}", entry.origin()), + adapter: view + .session() + .port() + .unwrap_or("unknown-adapter") + .to_owned(), + link_settings: format!( + "baud={} parity={:?} data={:?} stop={:?} slave={} timeout_ms={} rs485={:?}", + link.current.baud_rate.get(), + link.current.parity, + link.current.data_bits, + link.current.stop_bits, + link.current.slave_id.get(), + link.current.response_timeout.as_millis(), + link.current.rs485_mode, + ), + drive_state: DriveState::Unknown, + started_at: now, + finished_at: now, + }) + } +} + +fn utc_now() -> UtcTimestamp { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| i128::try_from(duration.as_nanos()).unwrap_or(i128::MAX)) + .unwrap_or(0); + UtcTimestamp::from_unix_nanos(nanos) +} + +#[derive(Clone, Debug, Default)] +pub struct ApplicationView { + inner: legacy::ApplicationView, + backup: BackupRestoreView, +} + +impl ApplicationView { + #[must_use] + pub fn active_profile_id(&self) -> Option<&str> { + self.inner.active_profile_id() + } + + #[must_use] + pub const fn active_session(&self) -> Option { + self.inner.active_session() + } + + #[must_use] + pub fn registry_profile_ids(&self) -> &[String] { + self.inner.registry_profile_ids() + } + + #[must_use] + pub const fn session(&self) -> &SessionView { + self.inner.session() + } + + #[must_use] + pub const fn connection(&self) -> &ConnectionWizardView { + self.inner.connection() + } + + #[must_use] + pub const fn monitoring(&self) -> &MonitoringView { + self.inner.monitoring() + } + + #[must_use] + pub const fn parameters(&self) -> &ParameterBrowserView { + self.inner.parameters() + } + + #[must_use] + pub const fn faults(&self) -> &FaultTimelineView { + self.inner.faults() + } + + #[must_use] + pub const fn backup(&self) -> &BackupRestoreView { + &self.backup + } +} diff --git a/crates/lantern-storage/src/backup_catalog.rs b/crates/lantern-storage/src/backup_catalog.rs new file mode 100644 index 00000000..368328ca --- /dev/null +++ b/crates/lantern-storage/src/backup_catalog.rs @@ -0,0 +1,57 @@ +use std::{ + fs, io, + path::{Path, PathBuf}, +}; + +use crate::BACKUP_SUFFIX; + +/// Lists regular backup files in deterministic path order. Symlinks and non-files are omitted; +/// actual backup parsing remains bounded and validated by `read_backup`. +pub fn list_backup_files(directory: &Path) -> io::Result> { + if !directory.exists() { + return Ok(Vec::new()); + } + + let mut paths = Vec::new(); + for entry in fs::read_dir(directory)? { + let entry = entry?; + if !entry.file_type()?.is_file() { + continue; + } + let path = entry.path(); + if path + .file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.ends_with(BACKUP_SUFFIX)) + { + paths.push(path); + } + } + paths.sort(); + Ok(paths) +} + +#[cfg(test)] +mod tests { + use std::{fs, os::unix::fs::symlink}; + + use tempfile::tempdir; + + use super::list_backup_files; + + #[test] + fn catalog_is_sorted_and_ignores_non_backup_and_symlink_entries() { + let directory = tempdir().expect("tempdir"); + let a = directory.path().join("a.vfdlantern-backup.json"); + let b = directory.path().join("b.vfdlantern-backup.json"); + fs::write(&b, b"b").expect("b"); + fs::write(&a, b"a").expect("a"); + fs::write(directory.path().join("notes.txt"), b"ignore").expect("notes"); + symlink(&a, directory.path().join("linked.vfdlantern-backup.json")).expect("symlink"); + + assert_eq!( + list_backup_files(directory.path()).expect("catalog"), + vec![a, b] + ); + } +} diff --git a/crates/lantern-storage/src/lib.rs b/crates/lantern-storage/src/lib.rs index ec7afb70..275f26cf 100644 --- a/crates/lantern-storage/src/lib.rs +++ b/crates/lantern-storage/src/lib.rs @@ -6,6 +6,7 @@ mod artifacts; mod atomic; mod audit; mod backup; +mod backup_catalog; mod csv_lifecycle; mod csv_writer; mod diagnostics_bundle; @@ -30,6 +31,7 @@ pub use backup::{ BACKUP_SCHEMA_VERSION, BACKUP_SUFFIX, BackupEnvelopeV1, BackupPayloadV1, BackupStorageError, MAX_BACKUP_FILE_BYTES, MAX_BACKUP_VALUES, decode_backup, read_backup, write_backup, }; +pub use backup_catalog::list_backup_files; pub use csv_lifecycle::{CsvLoggingCoordinator, CsvLoggingLifecycleState}; pub use csv_writer::{ CSV_HEADER, CSV_SCHEMA_VERSION, CsvWriterActor, CsvWriterHandle, CsvWriterStart, diff --git a/crates/lantern-tui/src/backup_keymap.rs b/crates/lantern-tui/src/backup_keymap.rs new file mode 100644 index 00000000..c8f9931f --- /dev/null +++ b/crates/lantern-tui/src/backup_keymap.rs @@ -0,0 +1,62 @@ +use crossterm::event::{KeyCode, KeyEvent, KeyEventKind, KeyModifiers}; +use lantern_app::{ApplicationAction, ApplicationView, BackupAction}; + +use crate::{ConnectionEdit, MappedAction, UiAction, UiState}; + +#[must_use] +pub fn map_backup_key(ui: &UiState, view: &ApplicationView, key: KeyEvent) -> Option { + if !matches!(key.kind, KeyEventKind::Press | KeyEventKind::Repeat) { + return None; + } + if key.modifiers.contains(KeyModifiers::CONTROL) && key.code == KeyCode::Char('c') { + return None; + } + + if ui.connection_edit == Some(ConnectionEdit::WriteConfirmation) { + return match key.code { + KeyCode::Esc => Some(MappedAction::Ui(UiAction::CancelEdit)), + KeyCode::Enter => Some(MappedAction::Combined { + ui: UiAction::CancelEdit, + application: Box::new(ApplicationAction::Backup(Box::new( + BackupAction::ConfirmRestore { + operator_text: ui.form.value().to_owned(), + }, + ))), + }), + KeyCode::Backspace => Some(MappedAction::Ui(UiAction::Backspace)), + KeyCode::Char(character) => Some(MappedAction::Ui(UiAction::InputChar(character))), + _ => None, + }; + } + + match key.code { + KeyCode::Char('b') => Some(MappedAction::Application(Box::new( + ApplicationAction::Backup(Box::new(BackupAction::Capture)), + ))), + KeyCode::Char('r') => Some(MappedAction::Application(Box::new( + ApplicationAction::Backup(Box::new(BackupAction::RefreshCatalog)), + ))), + KeyCode::Char('x') => Some(MappedAction::Application(Box::new( + ApplicationAction::Backup(Box::new(BackupAction::ClearSource)), + ))), + KeyCode::Char('p') => Some(MappedAction::Application(Box::new( + ApplicationAction::Backup(Box::new(BackupAction::PrepareRestore)), + ))), + KeyCode::Char('c') if view.backup().prepared_plan.is_some() => { + Some(MappedAction::Ui(UiAction::BeginWriteConfirmation)) + } + KeyCode::Enter => view + .backup() + .catalog + .get(ui.selected_index) + .cloned() + .map(|path| { + MappedAction::Application(Box::new(ApplicationAction::Backup(Box::new( + BackupAction::SelectSource(path), + )))) + }), + KeyCode::Char('j') | KeyCode::Down => Some(MappedAction::Ui(UiAction::SelectionNext)), + KeyCode::Char('k') | KeyCode::Up => Some(MappedAction::Ui(UiAction::SelectionPrevious)), + _ => None, + } +} diff --git a/crates/lantern-tui/src/backup_render.rs b/crates/lantern-tui/src/backup_render.rs new file mode 100644 index 00000000..20826a58 --- /dev/null +++ b/crates/lantern-tui/src/backup_render.rs @@ -0,0 +1,151 @@ +use lantern_app::ApplicationView; +use ratatui::{ + Frame, + layout::Rect, + text::{Line, Text}, + widgets::{Block, Paragraph, Wrap}, +}; + +use crate::{ConnectionEdit, Theme, UiState}; + +pub fn render_backup_screen( + frame: &mut Frame<'_>, + area: Rect, + view: &ApplicationView, + ui: &UiState, + theme: Theme, +) { + let backup = view.backup(); + let mut lines = vec![ + Line::from( + "b capture | r refresh files | Enter select source | p prepare restore | c confirm | x clear source", + ), + Line::from( + "Restore remains gated by Verified + trust + Armed + healthy audit + exact confirmation + permit.", + ), + Line::from(""), + ]; + + if view.active_session().is_none() { + lines.push(Line::from( + "No Verified session: catalog/source inspection is available; capture/restore is blocked.", + )); + } else { + lines.push(Line::from(format!( + "session={:?} profile={} operation={:?} authorization={:?} audit={:?}", + view.active_session().map(lantern_app::SessionId::get), + view.session().verified_profile_id().unwrap_or("—"), + view.session().operation(), + view.session().authorization(), + view.session().audit_health(), + ))); + } + + if let Some(status) = &backup.status { + lines.push(Line::from(format!("STATUS: {status}"))); + } + if let Some(error) = &backup.error { + lines.push(Line::from(format!("ERROR: {error}"))); + } + + lines.push(Line::from("")); + if let Some(last) = &backup.last_capture { + lines.push(Line::from(format!( + "Last capture: {} id={} complete={} values={} errors={}", + last.path.display(), + last.backup_id, + last.complete, + last.values, + last.errors, + ))); + } + if let Some(source) = &backup.source { + lines.push(Line::from(format!( + "Restore source: {} id={} complete={} profile={} hash={}", + source.path.display(), + source.backup_id, + source.complete, + source.profile_id, + source.profile_hash, + ))); + } else { + lines.push(Line::from("Restore source: —")); + } + if let Some(current) = &backup.pre_restore { + lines.push(Line::from(format!( + "Fresh pre-restore backup: {} id={} complete={} values={}", + current.path.display(), + current.backup_id, + current.complete, + current.values, + ))); + } + + lines.push(Line::from("")); + lines.push(Line::from("Stored backups:")); + if backup.catalog.is_empty() { + lines.push(Line::from(" no stored backup files")); + } + for (index, path) in backup.catalog.iter().enumerate() { + let marker = if index == ui.selected_index { ">" } else { " " }; + lines.push(Line::from(format!("{marker} {}", path.display()))); + } + + if !backup.diff.is_empty() { + lines.push(Line::from("")); + lines.push(Line::from(format!( + "Semantic diff ({} entries):", + backup.diff.len() + ))); + for entry in backup.diff.iter().take(64) { + lines.push(Line::from(format!( + " {} {:?}", + entry.parameter_id, entry.status + ))); + } + if backup.diff.len() > 64 { + lines.push(Line::from(format!( + " … {} additional diff entries", + backup.diff.len() - 64 + ))); + } + } + + if let Some(plan) = &backup.prepared_plan { + lines.push(Line::from("")); + lines.push(Line::from(format!( + "PREPARED RESTORE: steps={} skipped={} plan_hash={}", + plan.steps.len(), + plan.skipped, + plan.plan_hash, + ))); + for step in plan.steps.iter().take(32) { + lines.push(Line::from(format!( + " #{} {} old={} target={}", + step.index, step.parameter_id, step.expected_old, step.target + ))); + } + lines.push(Line::from(format!( + "Exact confirmation required: {}", + plan.challenge + ))); + if ui.connection_edit == Some(ConnectionEdit::WriteConfirmation) { + lines.push(Line::from(format!("Confirmation: {}_", ui.form.value()))); + lines.push(Line::from( + "Enter submits exact text; Esc cancels without write.", + )); + } else { + lines.push(Line::from( + "Press c to enter the exact confirmation challenge.", + )); + } + } + + let scroll = u16::try_from(ui.scroll_offset).unwrap_or(u16::MAX); + let paragraph = Paragraph::new(Text::from(lines)) + .block(Block::bordered().title(" Backup / Diff / Restore ")) + .wrap(Wrap { trim: false }) + .scroll((scroll, 0)) + .style(theme.muted()); + frame.render_widget(paragraph, area); +} diff --git a/crates/lantern-tui/src/input.rs b/crates/lantern-tui/src/input.rs index f22d2eee..e962f679 100644 --- a/crates/lantern-tui/src/input.rs +++ b/crates/lantern-tui/src/input.rs @@ -4,7 +4,7 @@ use crossterm::event::{Event, EventStream}; use futures_util::StreamExt; use lantern_app::ApplicationView; -use crate::{MappedAction, UiAction, UiState, map_key}; +use crate::{MappedAction, Screen, UiAction, UiState, backup_keymap::map_backup_key, map_key}; pub struct InputReader { events: EventStream, @@ -29,6 +29,11 @@ impl InputReader { })??; match event { Event::Key(key) => { + if ui.screen == Screen::Backup + && let Some(action) = map_backup_key(ui, view, key) + { + return Ok(action); + } if let Some(action) = map_key(ui, view, key) { return Ok(action); } diff --git a/crates/lantern-tui/src/lib.rs b/crates/lantern-tui/src/lib.rs index 1d35c887..0ea425b1 100644 --- a/crates/lantern-tui/src/lib.rs +++ b/crates/lantern-tui/src/lib.rs @@ -2,6 +2,8 @@ #![forbid(unsafe_code)] +mod backup_keymap; +mod backup_render; mod fault_keymap; mod fault_render; mod fault_state; @@ -40,6 +42,7 @@ use ratatui::{ }; use crate::{ + backup_render::render_backup_screen, screens::render_screen, widgets::{render_footer, render_header, render_modal, render_navigation, render_too_small}, }; @@ -65,7 +68,11 @@ pub fn render(frame: &mut Frame<'_>, view: &ApplicationView, ui: &UiState, theme render_header(frame, header, view, theme); render_navigation(frame, navigation, ui, theme); - render_screen(frame, content, view, ui, theme); + if ui.screen == Screen::Backup { + render_backup_screen(frame, content, view, ui, theme); + } else { + render_screen(frame, content, view, ui, theme); + } render_footer(frame, footer, theme); if let Some(modal) = &ui.modal { diff --git a/crates/vfd-lantern/src/connection_runtime.rs b/crates/vfd-lantern/src/connection_runtime.rs index 382100e7..29dcd377 100644 --- a/crates/vfd-lantern/src/connection_runtime.rs +++ b/crates/vfd-lantern/src/connection_runtime.rs @@ -39,6 +39,7 @@ struct RuntimeState { pub struct TuiRuntimePaths { diagnostics_directory: PathBuf, fault_report_directory: PathBuf, + backup_directory: PathBuf, csv_directory: PathBuf, session_runtime_directory: PathBuf, audit_directory: PathBuf, @@ -50,6 +51,7 @@ impl TuiRuntimePaths { pub fn new( diagnostics_directory: PathBuf, fault_report_directory: PathBuf, + backup_directory: PathBuf, csv_directory: PathBuf, session_runtime_directory: PathBuf, audit_directory: PathBuf, @@ -58,6 +60,7 @@ impl TuiRuntimePaths { Self { diagnostics_directory, fault_report_directory, + backup_directory, csv_directory, session_runtime_directory, audit_directory, @@ -92,6 +95,7 @@ impl TuiEffectRunner { registry, paths.audit_directory.clone(), paths.profile_trust_store.clone(), + paths.backup_directory.clone(), settings.process_writes_enabled, ); let monitoring = MonitoringRuntime::new( @@ -382,6 +386,7 @@ impl EffectRunner for TuiEffectRunner { ApplicationEffect::Faults(effect) => self.execute_fault(effect), ApplicationEffect::Write(effect) => self.write.execute(effect), ApplicationEffect::Session(effect) => self.execute_session(effect), + ApplicationEffect::Backup(effect) => self.write.execute_backup(*effect), } } } diff --git a/crates/vfd-lantern/src/main.rs b/crates/vfd-lantern/src/main.rs index fffed7ec..87d2cfc4 100644 --- a/crates/vfd-lantern/src/main.rs +++ b/crates/vfd-lantern/src/main.rs @@ -22,8 +22,8 @@ use std::{ use anyhow::Result; use clap::Parser; use lantern_app::{ - ApplicationAction, ApplicationRuntime, ApplicationState, CliSettingsOverrides, ColorMode, - ConnectionAction, ParameterAction, PortDiscoveryPort, PortEvent, PortEventReceiver, + ApplicationAction, ApplicationRuntime, ApplicationState, BackupAction, CliSettingsOverrides, + ColorMode, ConnectionAction, ParameterAction, PortDiscoveryPort, PortEvent, PortEventReceiver, ProfileRegistry, SessionInput, SessionPhaseView, SettingsLoader, ValidatedSettings, }; use lantern_storage::{ @@ -150,12 +150,13 @@ async fn run_tui(settings: &ValidatedSettings, paths: &AppPaths) -> Result<()> { install_terminal_panic_hook(Arc::clone(&terminal_guard), paths.panic_directory.clone()); let (action_tx, mut action_rx) = mpsc::unbounded_channel(); - let state = ApplicationState::with_registry_and_suggestions( + let mut state = ApplicationState::with_registry_and_suggestions( Arc::clone(®istry), settings.process_writes_enabled, settings.suggested_device.clone(), settings.suggested_slave, ); + state.set_build_id(profile_commands::embedded_manifest()?.build_id); let runner = TuiEffectRunner::new( Arc::clone(&terminal_guard), action_tx, @@ -164,6 +165,7 @@ async fn run_tui(settings: &ValidatedSettings, paths: &AppPaths) -> Result<()> { TuiRuntimePaths::new( paths.diagnostics_directory.clone(), paths.fault_report_directory.clone(), + paths.backup_directory.clone(), paths.csv_directory.clone(), paths.session_runtime_directory.clone(), paths.audit_directory.clone(), @@ -180,6 +182,9 @@ async fn run_tui(settings: &ValidatedSettings, paths: &AppPaths) -> Result<()> { application.dispatch(ApplicationAction::Connection( ConnectionAction::RefreshPorts, ))?; + application.dispatch(ApplicationAction::Backup(Box::new( + BackupAction::RefreshCatalog, + )))?; terminal.draw(&application.state().view(), &ui)?; let frame_interval = Duration::from_millis(1_000 / u64::from(settings.render_fps)); diff --git a/crates/vfd-lantern/src/write_runtime.rs b/crates/vfd-lantern/src/write_runtime.rs index 2a5013a9..e7b73f54 100644 --- a/crates/vfd-lantern/src/write_runtime.rs +++ b/crates/vfd-lantern/src/write_runtime.rs @@ -1,28 +1,34 @@ use std::{ - path::PathBuf, + path::{Path, PathBuf}, sync::{Arc, Mutex, MutexGuard}, time::Instant, }; use lantern_app::{ - ApplicationAction, ApplicationEffectError, AuditPort, ClockPort, DecisionOutcome, - DeviceFingerprint, DeviceWriteOutcome, DriveState, OperationId, ParameterAction, PlanId, - ProfileRegistry, ProfileTrustPort, ReadBusPort, SessionControlError, SessionControlPort, - SessionId, SessionInput, SlaveId, WriteBusPort, WriteCoordinator, WriteCoordinatorConfig, - WriteEffect, WriteOutcome, WriteSessionSnapshot, + ApplicationAction, ApplicationEffectError, AuditPort, BackupAction, BackupCoordinator, + BackupEffect, ClockPort, DecisionOutcome, DeviceFingerprint, DeviceWriteOutcome, DriveState, + OperationId, ParameterAction, PlanId, PreparedRestoreBundle, ProfileRegistry, ProfileTrustPort, + ReadBusPort, RestoreExecutionSummary, SessionControlError, SessionControlPort, SessionId, + SessionInput, SlaveId, StoredBackup, WriteBusPort, WriteCoordinator, WriteCoordinatorConfig, + WriteEffect, WriteOutcome, WriteSessionSnapshot, semantic_backup_diff, +}; +use lantern_storage::{ + BACKUP_SUFFIX, FilesystemAuditPort, RuntimeProfileTrust, list_backup_files, read_backup, + write_backup, }; -use lantern_storage::{FilesystemAuditPort, RuntimeProfileTrust}; use lantern_transport::BusActorHandle; use tokio::sync::{Mutex as AsyncMutex, mpsc}; #[derive(Clone)] pub struct ProductionWriteRuntime { coordinator: Arc>>, + backup: Arc>>, session: Arc, audit: Option>, trust: Option>, clock: Arc, config: WriteCoordinatorConfig, + backup_directory: PathBuf, action_tx: mpsc::UnboundedSender, } @@ -33,6 +39,7 @@ impl ProductionWriteRuntime { registry: Arc, audit_directory: PathBuf, trust_store_path: PathBuf, + backup_directory: PathBuf, process_writes_enabled: bool, ) -> Self { let audit: Option> = match FilesystemAuditPort::new(audit_directory) { @@ -48,18 +55,42 @@ impl ProductionWriteRuntime { registry, trust_store_path, ))); - Self::from_adapters(action_tx, audit, trust, process_writes_enabled) + Self::from_adapters_with_backup_directory( + action_tx, + audit, + trust, + backup_directory, + process_writes_enabled, + ) } + #[cfg(test)] fn from_adapters( action_tx: mpsc::UnboundedSender, audit: Option>, trust: Option>, process_writes_enabled: bool, + ) -> Self { + Self::from_adapters_with_backup_directory( + action_tx, + audit, + trust, + PathBuf::from("."), + process_writes_enabled, + ) + } + + fn from_adapters_with_backup_directory( + action_tx: mpsc::UnboundedSender, + audit: Option>, + trust: Option>, + backup_directory: PathBuf, + process_writes_enabled: bool, ) -> Self { let session = Arc::new(RuntimeSessionControl::new(action_tx.clone())); Self { coordinator: Arc::new(AsyncMutex::new(None)), + backup: Arc::new(AsyncMutex::new(None)), session, audit, trust, @@ -68,6 +99,7 @@ impl ProductionWriteRuntime { process_writes_enabled, ..WriteCoordinatorConfig::default() }, + backup_directory, action_tx, } } @@ -79,6 +111,27 @@ impl ProductionWriteRuntime { } async fn attach_ports(&self, read_bus: Arc, write_bus: Arc) { + let clock: Arc = self.clock.clone(); + let session: Arc = self.session.clone(); + + if let Some(trust) = self.trust.clone() { + match BackupCoordinator::new( + Arc::clone(&read_bus), + trust, + Arc::clone(&clock), + Arc::clone(&session), + self.config.request_timeout, + ) { + Ok(coordinator) => *self.backup.lock().await = Some(coordinator), + Err(error) => { + eprintln!("backup coordinator unavailable: {error}"); + *self.backup.lock().await = None; + } + } + } else { + *self.backup.lock().await = None; + } + let Some(audit) = self.audit.clone() else { *self.coordinator.lock().await = None; return; @@ -87,8 +140,6 @@ impl ProductionWriteRuntime { *self.coordinator.lock().await = None; return; }; - let clock: Arc = self.clock.clone(); - let session: Arc = self.session.clone(); match WriteCoordinator::new( read_bus, write_bus, @@ -184,6 +235,169 @@ impl ProductionWriteRuntime { } } } + + pub fn execute_backup(&self, effect: BackupEffect) -> Result<(), ApplicationEffectError> { + match effect { + BackupEffect::RefreshCatalog => { + let result = backup_catalog(&self.backup_directory); + send_backup_action(&self.action_tx, BackupAction::CatalogRefreshed(result)) + } + BackupEffect::LoadSource { path } => { + let sender = self.action_tx.clone(); + tokio::task::spawn_blocking(move || { + let result = read_backup(&path).map_err(|error| error.to_string()); + let _ = sender.send(ApplicationAction::Backup(Box::new( + BackupAction::SourceLoaded { path, result }, + ))); + }); + Ok(()) + } + BackupEffect::Capture { context } => { + let backup = Arc::clone(&self.backup); + let directory = self.backup_directory.clone(); + let sender = self.action_tx.clone(); + tokio::spawn(async move { + let result: Result = async { + let snapshot = backup + .lock() + .await + .as_mut() + .ok_or_else(|| { + "backup capability unavailable: no verified bus/trust composition" + .to_owned() + })? + .capture(context) + .await + .map_err(|error| error.to_string())?; + persist_backup(&directory, snapshot) + } + .await; + let _ = sender.send(ApplicationAction::Backup(Box::new( + BackupAction::Captured(result), + ))); + }); + Ok(()) + } + BackupEffect::PrepareRestore { source, context } => { + let backup = Arc::clone(&self.backup); + let coordinator = Arc::clone(&self.coordinator); + let directory = self.backup_directory.clone(); + let sender = self.action_tx.clone(); + tokio::spawn(async move { + let result: Result = async { + let current = backup + .lock() + .await + .as_mut() + .ok_or_else(|| { + "backup capability unavailable before restore".to_owned() + })? + .capture(context) + .await + .map_err(|error| error.to_string())?; + let pre_restore = persist_backup(&directory, current)?; + let diff = semantic_backup_diff(source.as_ref(), &pre_restore.snapshot, None); + let plan = coordinator + .lock() + .await + .as_mut() + .ok_or_else(|| { + "restore capability unavailable: write/audit/trust composition is incomplete" + .to_owned() + })? + .prepare_restore_plan(source.as_ref(), &pre_restore.snapshot) + .await + .map_err(|error| error.to_string())?; + Ok(PreparedRestoreBundle { + pre_restore, + diff, + plan, + }) + } + .await; + let _ = sender.send(ApplicationAction::Backup(Box::new( + BackupAction::RestorePrepared(Box::new(result)), + ))); + }); + Ok(()) + } + BackupEffect::ExecuteRestore { plan, confirmation } => { + let coordinator = Arc::clone(&self.coordinator); + let sender = self.action_tx.clone(); + tokio::spawn(async move { + let result: Result = async { + let total = plan.steps().len(); + let mut guard = coordinator.lock().await; + let coordinator = guard.as_mut().ok_or_else(|| { + "restore capability unavailable: write/audit/trust composition is incomplete" + .to_owned() + })?; + let mut permit = coordinator + .begin_restore(plan, confirmation) + .await + .map_err(|error| error.to_string())?; + let mut verified_steps = 0_usize; + for index in 0..total { + let outcome = coordinator + .execute_restore_step(&mut permit, index) + .await + .map_err(|error| error.to_string())?; + if outcome == DeviceWriteOutcome::Verified { + verified_steps = verified_steps.saturating_add(1); + continue; + } + return Ok(RestoreExecutionSummary { + attempted_steps: index.saturating_add(1), + verified_steps, + terminal_outcome: Some(outcome), + }); + } + coordinator + .finish_restore(permit) + .await + .map_err(|error| error.to_string())?; + Ok(RestoreExecutionSummary { + attempted_steps: total, + verified_steps, + terminal_outcome: None, + }) + } + .await; + let _ = sender.send(ApplicationAction::Backup(Box::new( + BackupAction::RestoreCompleted(result), + ))); + }); + Ok(()) + } + } + } +} + +fn send_backup_action( + sender: &mpsc::UnboundedSender, + action: BackupAction, +) -> Result<(), ApplicationEffectError> { + sender + .send(ApplicationAction::Backup(Box::new(action))) + .map_err(|_| ApplicationEffectError("application action channel closed".to_owned())) +} + +fn backup_catalog(directory: &Path) -> Result, String> { + list_backup_files(directory).map_err(|error| error.to_string()) +} + +fn persist_backup( + directory: &Path, + snapshot: lantern_app::BackupSnapshot, +) -> Result { + let path = directory.join(format!( + "backup-{}-{}{}", + snapshot.backup_id.get(), + snapshot.finished_at.as_unix_nanos(), + BACKUP_SUFFIX + )); + write_backup(&path, &snapshot).map_err(|error| error.to_string())?; + Ok(StoredBackup { path, snapshot }) } struct RuntimeWriteClock { @@ -204,8 +418,16 @@ impl ClockPort for RuntimeWriteClock { } } +#[derive(Clone, Debug)] +struct RuntimeRestoreState { + operation_id: OperationId, + plan_hash: String, + next_index: usize, +} + struct RuntimeSessionControl { snapshot: Mutex, + restore: Mutex>, action_tx: mpsc::UnboundedSender, } @@ -213,11 +435,15 @@ impl RuntimeSessionControl { fn new(action_tx: mpsc::UnboundedSender) -> Self { Self { snapshot: Mutex::new(unavailable_snapshot()), + restore: Mutex::new(None), action_tx, } } fn sync(&self, snapshot: WriteSessionSnapshot) { + if snapshot.operation_idle { + *lock_restore(&self.restore) = None; + } *lock_snapshot(&self.snapshot) = snapshot; } } @@ -285,6 +511,156 @@ impl SessionControlPort for RuntimeSessionControl { })); } + fn begin_restore( + &self, + operation_id: OperationId, + plan_hash: &str, + ) -> Result<(), SessionControlError> { + let mut snapshot = lock_snapshot(&self.snapshot); + let mut restore = lock_restore(&self.restore); + if !snapshot.connected + || !snapshot.armed + || !snapshot.audit_healthy + || !snapshot.operation_idle + || restore.is_some() + || plan_hash.is_empty() + { + return Err(SessionControlError::PreconditionChanged); + } + snapshot.operation_idle = false; + snapshot.guard_revision = snapshot.guard_revision.saturating_add(1); + *restore = Some(RuntimeRestoreState { + operation_id, + plan_hash: plan_hash.to_owned(), + next_index: 0, + }); + if self + .action_tx + .send(ApplicationAction::Session(SessionInput::RestoreStarted { + operation_id, + plan_hash: plan_hash.to_owned(), + })) + .is_err() + { + *restore = None; + snapshot.operation_idle = true; + snapshot.armed = false; + snapshot.guard_revision = snapshot.guard_revision.saturating_add(1); + return Err(SessionControlError::Other( + "application session channel closed while starting restore".to_owned(), + )); + } + Ok(()) + } + + fn restore_matches( + &self, + operation_id: OperationId, + plan_hash: &str, + next_index: usize, + ) -> bool { + let snapshot = lock_snapshot(&self.snapshot); + let restore = lock_restore(&self.restore); + snapshot.connected + && snapshot.armed + && snapshot.audit_healthy + && !snapshot.operation_idle + && restore.as_ref().is_some_and(|state| { + state.operation_id == operation_id + && state.plan_hash == plan_hash + && state.next_index == next_index + }) + } + + fn advance_restore( + &self, + operation_id: OperationId, + plan_hash: &str, + next_index: usize, + ) -> Result<(), SessionControlError> { + let mut snapshot = lock_snapshot(&self.snapshot); + let mut restore = lock_restore(&self.restore); + let Some(state) = restore.as_mut() else { + return Err(SessionControlError::PreconditionChanged); + }; + if !snapshot.connected + || !snapshot.armed + || !snapshot.audit_healthy + || snapshot.operation_idle + || state.operation_id != operation_id + || state.plan_hash != plan_hash + || next_index != state.next_index.saturating_add(1) + { + return Err(SessionControlError::PreconditionChanged); + } + state.next_index = next_index; + snapshot.guard_revision = snapshot.guard_revision.saturating_add(1); + if self + .action_tx + .send(ApplicationAction::Session(SessionInput::RestoreAdvanced { + next_index, + })) + .is_err() + { + return Err(SessionControlError::Other( + "application session channel closed while advancing restore".to_owned(), + )); + } + Ok(()) + } + + fn finish_restore( + &self, + operation_id: OperationId, + plan_hash: &str, + ) -> Result<(), SessionControlError> { + let mut snapshot = lock_snapshot(&self.snapshot); + let mut restore = lock_restore(&self.restore); + let matches = restore.as_ref().is_some_and(|state| { + state.operation_id == operation_id && state.plan_hash == plan_hash + }); + if !matches || snapshot.operation_idle { + return Err(SessionControlError::PreconditionChanged); + } + *restore = None; + snapshot.operation_idle = true; + snapshot.armed = false; + snapshot.guard_revision = snapshot.guard_revision.saturating_add(1); + self.action_tx + .send(ApplicationAction::Session(SessionInput::RestoreFinished)) + .map_err(|_| { + SessionControlError::Other( + "application session channel closed while finishing restore".to_owned(), + ) + }) + } + + fn abort_restore( + &self, + operation_id: OperationId, + plan_hash: &str, + ) -> Result<(), SessionControlError> { + let mut snapshot = lock_snapshot(&self.snapshot); + let mut restore = lock_restore(&self.restore); + let matches = restore.as_ref().is_some_and(|state| { + state.operation_id == operation_id && state.plan_hash == plan_hash + }); + if !matches { + return Err(SessionControlError::PreconditionChanged); + } + *restore = None; + snapshot.operation_idle = true; + snapshot.armed = false; + snapshot.guard_revision = snapshot.guard_revision.saturating_add(1); + self.action_tx + .send(ApplicationAction::Session(SessionInput::RestoreAborted)) + .map_err(|_| { + SessionControlError::Other( + "application session channel closed while aborting restore".to_owned(), + ) + }) + } + fn disarm(&self) { { let mut snapshot = lock_snapshot(&self.snapshot); @@ -304,6 +680,7 @@ impl SessionControlPort for RuntimeSessionControl { snapshot.operation_idle = true; snapshot.guard_revision = snapshot.guard_revision.saturating_add(1); } + *lock_restore(&self.restore) = None; let _ = self.action_tx.send(ApplicationAction::Session( SessionInput::AuditPersistenceFailed { cause: "durable write audit failed".to_owned(), @@ -323,6 +700,14 @@ fn lock_snapshot(snapshot: &Mutex) -> MutexGuard<'_, Write .unwrap_or_else(std::sync::PoisonError::into_inner) } +fn lock_restore( + restore: &Mutex>, +) -> MutexGuard<'_, Option> { + restore + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + fn unavailable_snapshot() -> WriteSessionSnapshot { WriteSessionSnapshot { session_id: SessionId::new(0), @@ -412,6 +797,7 @@ mod tests { let runtime = ProductionWriteRuntime::from_adapters(tx, Some(audit), None, true); let bus = attach_counting_bus(&runtime).await; assert!(runtime.coordinator.lock().await.is_none()); + assert!(runtime.backup.lock().await.is_none()); assert_eq!(bus.writes.load(Ordering::SeqCst), 0); } @@ -423,6 +809,7 @@ mod tests { ProductionWriteRuntime::from_adapters(tx, Some(audit), Some(trust_adapter()), true); let bus = attach_counting_bus(&runtime).await; assert!(runtime.coordinator.lock().await.is_some()); + assert!(runtime.backup.lock().await.is_some()); assert_eq!(bus.writes.load(Ordering::SeqCst), 0); } }