diff --git a/.gitignore b/.gitignore index 04ae708..e5005c1 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,7 @@ /target __pycache__/ *.pyc +pchi_governance_key.hex +pchi_governance_log.jsonl +pchi_scene_state.json +pchi_scene_state.json.tmp diff --git a/primeswarm-pchi/src/conductor.rs b/primeswarm-pchi/src/conductor.rs index 81ef38e..eb1297d 100644 --- a/primeswarm-pchi/src/conductor.rs +++ b/primeswarm-pchi/src/conductor.rs @@ -23,6 +23,12 @@ pub struct PCHIConductor { /// Equilibrium threshold equilibrium_threshold: f64, + + /// Where to periodically snapshot scene state, if persistence is + /// enabled (`None` by default — see `with_state_file`). A crash + /// restart with this unset starts from a blank `SceneState`, same as + /// every version of this conductor before this field existed. + state_file: Option, } impl PCHIConductor { @@ -53,6 +59,7 @@ impl PCHIConductor { governance: Arc::new(GovernanceLog::open(signing_key_path, governance_log_path)?), transport: TransportLayer::new(), equilibrium_threshold, + state_file: None, }) } @@ -70,9 +77,38 @@ impl PCHIConductor { governance: Arc::new(GovernanceLog::open(signing_key_path, governance_log_path)?), transport: TransportLayer::new(), equilibrium_threshold, + state_file: None, }) } + /// Enable periodic scene-state persistence to `path`, loading an + /// existing snapshot from it right now if one's there — so a restart + /// resumes from the last snapshot instead of starting blank. Chain + /// this onto `new`/`with_governance_paths`/`with_rules_file` before + /// wrapping the conductor in `Arc` and spawning its background loops; + /// `run_state_snapshot_loop` is what actually keeps writing to `path` + /// afterward. + /// + /// This is a resilience convenience, not the safety-critical + /// durability guarantee the governance log already has (that log + /// fsyncs every signed receipt as it's recorded) — losing scene + /// state on an ungraceful crash means tools and the conductor + /// disagree about "where were we," not that an unsafe state went + /// unlogged. + pub fn with_state_file(mut self, path: &str) -> Self { + match SceneState::load_snapshot(path) { + Some(loaded) => { + info!("Resumed scene state from {} (last updated {})", path, loaded.timestamp); + self.state = Arc::new(RwLock::new(loaded)); + } + None => { + info!("No existing scene state snapshot at {} — starting fresh", path); + } + } + self.state_file = Some(path.to_string()); + self + } + /// Process an incoming PCHI message pub async fn process_message(&self, message: PCHIMessage) -> Result<(), ConductorError> { // Validate the message @@ -285,6 +321,50 @@ impl PCHIConductor { pub async fn stop_transport(&mut self) -> Result<(), ConductorError> { self.transport.stop().await } + + /// Periodically snapshot scene state to disk, if `with_state_file` + /// enabled it — a no-op loop that returns immediately otherwise, so + /// it's always safe to spawn alongside the other background loops + /// regardless of whether persistence is configured. Skips the write + /// when nothing has changed since the last snapshot, compared via + /// `SceneState::timestamp` (already updated by every mutation), so + /// an idle conductor isn't rewriting an unchanged file every + /// interval. + pub async fn run_state_snapshot_loop(self: Arc, interval: std::time::Duration) { + let Some(path) = self.state_file.clone() else { + return; + }; + let mut last_snapshotted: Option> = None; + loop { + tokio::time::sleep(interval).await; + let (timestamp, snapshot) = { + let state = self.state.read().await; + (state.timestamp, state.clone()) + }; + if last_snapshotted == Some(timestamp) { + continue; + } + match snapshot.save_snapshot(&path) { + Ok(()) => last_snapshotted = Some(timestamp), + Err(e) => error!("Failed to write scene state snapshot to {}: {}", path, e), + } + } + } + + /// Snapshot scene state to disk right now, if persistence is + /// enabled. Called on graceful shutdown so a planned restart never + /// loses anything, regardless of where in `run_state_snapshot_loop`'s + /// interval the shutdown happened. + pub async fn snapshot_state_now(&self) { + let Some(path) = &self.state_file else { + return; + }; + let state = self.state.read().await; + match state.save_snapshot(path) { + Ok(()) => info!("Final scene state snapshot written to {}", path), + Err(e) => error!("Failed to write final scene state snapshot to {}: {}", path, e), + } + } } impl Default for PCHIConductor { @@ -488,6 +568,53 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + #[tokio::test] + async fn restart_resumes_scene_state_from_a_snapshot() { + let dir = std::env::temp_dir().join(format!("pchi-restart-test-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).unwrap(); + let state_path = dir.join("state.json"); + + // "Before restart": set a value, then snapshot the way graceful + // shutdown does. + let before = PCHIConductor::with_governance_paths( + 1e-12, + dir.join("key.hex").to_str().unwrap(), + dir.join("log.jsonl").to_str().unwrap(), + ) + .unwrap() + .with_state_file(state_path.to_str().unwrap()); + let msg = PCHIMessage::new( + MessageType::ControlParameter, + "test".to_string(), + Payload::ControlParameter(ControlParameterData { + target_id: "kraken_main".to_string(), + parameter: "wobbliness".to_string(), + value: serde_json::json!(0.42), + }), + ); + before.process_message(msg).await.unwrap(); + before.snapshot_state_now().await; + + // "After restart": a brand new conductor, same state file. + let after = PCHIConductor::with_governance_paths( + 1e-12, + dir.join("key2.hex").to_str().unwrap(), + dir.join("log2.jsonl").to_str().unwrap(), + ) + .unwrap() + .with_state_file(state_path.to_str().unwrap()); + + let resumed = after.get_state().await; + assert_eq!( + resumed.get_control_parameter("kraken_main.wobbliness"), + Some(0.42), + "a conductor pointed at the same state file must resume what the previous one set, \ + not start blank" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + #[tokio::test] async fn with_rules_file_rejects_a_missing_path_instead_of_silently_falling_back() { let dir = std::env::temp_dir().join(format!("pchi-rulesfile-missing-{}", uuid::Uuid::new_v4())); diff --git a/primeswarm-pchi/src/main.rs b/primeswarm-pchi/src/main.rs index ed5dcbc..cf2528f 100644 --- a/primeswarm-pchi/src/main.rs +++ b/primeswarm-pchi/src/main.rs @@ -17,6 +17,12 @@ struct Cli { /// same rules every fresh checkout has always run, unchanged. #[arg(long, env = "PCHI_RULES_FILE")] rules: Option, + + /// Where to periodically snapshot scene state, so a restart resumes + /// instead of starting blank. Always enabled, same as the governance + /// log's own default-local-file behavior. + #[arg(long, env = "PCHI_STATE_FILE", default_value = "pchi_scene_state.json")] + state_file: String, } #[tokio::main] @@ -46,7 +52,8 @@ async fn main() -> Result<(), Box> { }) } None => PCHIConductor::new(1e-12), - }; + } + .with_state_file(&cli.state_file); println!("PCHI Conductor started"); @@ -122,17 +129,31 @@ async fn main() -> Result<(), Box> { let ws_handle = tokio::spawn(conductor.clone().run_ws_accept_loop()); println!("Accepting governance-event WebSocket clients on ws://127.0.0.1:8889"); + // Periodically snapshot scene state so a restart resumes instead of + // starting blank. A no-op loop if persistence somehow ended up + // disabled, so always safe to spawn. + let snapshot_handle = tokio::spawn( + conductor.clone().run_state_snapshot_loop(std::time::Duration::from_secs(2)), + ); + println!("Persisting scene state to {} every 2s", cli.state_file); + println!("PCHI Conductor running. Press Ctrl+C to stop."); // Wait for shutdown signal tokio::signal::ctrl_c().await?; println!("Shutting down..."); + // One last snapshot before tearing down, so a planned restart loses + // nothing regardless of where in the snapshot loop's interval this + // landed. + conductor.snapshot_state_now().await; + // Stop background tasks (dropping the process closes the sockets; // stop_transport needs &mut self, which the Arc above gave up) web_handle.abort(); udp_handle.abort(); ws_handle.abort(); + snapshot_handle.abort(); println!("PCHI Conductor stopped"); diff --git a/primeswarm-pchi/src/state.rs b/primeswarm-pchi/src/state.rs index b2d9619..adbe554 100644 --- a/primeswarm-pchi/src/state.rs +++ b/primeswarm-pchi/src/state.rs @@ -143,6 +143,42 @@ impl SceneState { EquilibriumStatus::Disequilibrium { residual } }; } + + /// Atomically write this state to `path` as JSON: serialize, write to + /// a sibling temp file, then rename into place. The rename is what + /// makes it atomic — a crash mid-write leaves the temp file + /// incomplete but never touches `path` itself, so the file there is + /// always either the previous complete snapshot or the new one, + /// never a half-written one. + pub fn save_snapshot(&self, path: &str) -> std::io::Result<()> { + let json = serde_json::to_vec_pretty(self) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + let tmp_path = format!("{path}.tmp"); + std::fs::write(&tmp_path, json)?; + std::fs::rename(&tmp_path, path)?; + Ok(()) + } + + /// Load a previously saved snapshot, if one exists and parses. + /// Returns `None` rather than an error on a missing or corrupt file: + /// scene state is a resilience convenience, not the safety-critical + /// record (that's the governance log, which does fail hard on + /// corruption — see `governance.rs`) — a bad snapshot should log a + /// warning and start fresh, not keep the conductor from coming up. + pub fn load_snapshot(path: &str) -> Option { + let bytes = std::fs::read(path).ok()?; + match serde_json::from_slice(&bytes) { + Ok(state) => Some(state), + Err(e) => { + tracing::warn!( + "Scene state snapshot at {} is unreadable ({}), starting fresh", + path, + e + ); + None + } + } + } } impl Default for SceneState { @@ -150,3 +186,58 @@ impl Default for SceneState { Self::new() } } + +#[cfg(test)] +mod tests { + use super::*; + + fn temp_path(name: &str) -> std::path::PathBuf { + std::env::temp_dir().join(format!("pchi-state-test-{}-{}", name, uuid::Uuid::new_v4())) + } + + #[test] + fn save_then_load_round_trips_real_values() { + let path = temp_path("roundtrip"); + let mut state = SceneState::new(); + state.set_control_parameter("tentacle_system.total_curl".to_string(), 2.5); + state.update_musical_context(MusicalContext { + bpm: 128.0, + beat: 3, + kick: true, + snare: false, + section: "drop".to_string(), + }); + + state.save_snapshot(path.to_str().unwrap()).unwrap(); + let loaded = SceneState::load_snapshot(path.to_str().unwrap()).unwrap(); + + assert_eq!(loaded.get_control_parameter("tentacle_system.total_curl"), Some(2.5)); + assert_eq!(loaded.musical_context.unwrap().beat, 3); + assert_eq!(loaded.scene_id, state.scene_id, "identity, not just values, must survive a round trip"); + + let _ = std::fs::remove_file(&path); + } + + #[test] + fn save_leaves_no_temp_file_behind() { + let path = temp_path("no-temp-litter"); + SceneState::new().save_snapshot(path.to_str().unwrap()).unwrap(); + assert!(path.exists()); + assert!(!std::path::Path::new(&format!("{}.tmp", path.to_str().unwrap())).exists()); + let _ = std::fs::remove_file(&path); + } + + #[test] + fn load_missing_file_returns_none_not_an_error() { + let path = temp_path("does-not-exist"); + assert!(SceneState::load_snapshot(path.to_str().unwrap()).is_none()); + } + + #[test] + fn load_corrupt_file_returns_none_instead_of_panicking() { + let path = temp_path("corrupt"); + std::fs::write(&path, b"not valid json at all").unwrap(); + assert!(SceneState::load_snapshot(path.to_str().unwrap()).is_none()); + let _ = std::fs::remove_file(&path); + } +}