diff --git a/.sqlx/query-5f367f145e8aed4391af7a1b2b3de8f12ab61d44f3a664f89752b34338ff0b09.json b/.sqlx/query-5f367f145e8aed4391af7a1b2b3de8f12ab61d44f3a664f89752b34338ff0b09.json new file mode 100644 index 0000000..09ee45d --- /dev/null +++ b/.sqlx/query-5f367f145e8aed4391af7a1b2b3de8f12ab61d44f3a664f89752b34338ff0b09.json @@ -0,0 +1,26 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT payload AS \"payload: String\"\n FROM events\n WHERE aggregate_type = 'SchemaRegistry'\n AND aggregate_id = 'schema'\n ORDER BY sequence\n ", + "describe": { + "columns": [ + { + "name": "payload: String", + "ordinal": 0, + "type_info": "Null", + "origin": { + "Table": { + "table": "events", + "name": "payload" + } + } + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + false + ] + }, + "hash": "5f367f145e8aed4391af7a1b2b3de8f12ab61d44f3a664f89752b34338ff0b09" +} diff --git a/.sqlx/query-978a25f67d0b5d2b5a21c9f298d1b7ade60c52869bd382d8494520c7ba97cd42.json b/.sqlx/query-978a25f67d0b5d2b5a21c9f298d1b7ade60c52869bd382d8494520c7ba97cd42.json new file mode 100644 index 0000000..05a037e --- /dev/null +++ b/.sqlx/query-978a25f67d0b5d2b5a21c9f298d1b7ade60c52869bd382d8494520c7ba97cd42.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "\n DELETE FROM snapshots\n WHERE aggregate_type = ?1\n ", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "978a25f67d0b5d2b5a21c9f298d1b7ade60c52869bd382d8494520c7ba97cd42" +} diff --git a/crates/event-sorcery/src/engine.rs b/crates/event-sorcery/src/engine.rs index cacdae0..08ee82e 100644 --- a/crates/event-sorcery/src/engine.rs +++ b/crates/event-sorcery/src/engine.rs @@ -4,6 +4,7 @@ //! atomically commit aggregate events, snapshots, and durable job intent. use std::num::NonZeroUsize; +use std::sync::Arc; use cqrs_es::persist::{PersistenceError, ReplayStream, SerializedEvent, SerializedSnapshot}; use futures_util::TryStreamExt; @@ -14,6 +15,7 @@ use sqlx::{SqliteConnection, SqlitePool}; use crate::job::EnqueueRequest; use crate::job_sqlite::SqliteJobError; use crate::job_store::{ClaimDecision, ClaimOutcome, ClaimRead, LeaseRenewal}; +use crate::schema_registry::{ReconcileError, Reconciler, SchemaReconciliation, SchemaTarget}; const ENGINE_PAYLOAD_PROVENANCE_KEY: &str = "$event-sorcery-engine-owned"; const ENGINE_PAYLOAD_PROVENANCE: &str = "opaque-v1"; @@ -184,6 +186,7 @@ pub struct CommitRequest<'events> { #[derive(Clone)] pub struct Engine { pool: SqlitePool, + reconciler: Arc, } /// Failures produced by serialized engine operations. @@ -217,6 +220,8 @@ pub enum EngineError { Integer(#[from] std::num::TryFromIntError), #[error(transparent)] JobFlush(#[from] crate::job::JobStoreError), + #[error(transparent)] + Schema(#[from] ReconcileError), } /// An erased durable-job intent committed atomically with domain events. @@ -294,8 +299,14 @@ impl Engine { /// /// This constructor does not run migrations. Call [`Self::migrate`] before /// using storage operations when the pool's schema is not initialized. + /// + /// The schema registry's CQRS framework is built once here and reused by + /// every schema call, so one engine never holds more than one framework + /// instance for that aggregate. pub fn new(pool: SqlitePool) -> Self { - Self { pool } + let reconciler = Arc::new(Reconciler::new(pool.clone())); + + Self { pool, reconciler } } /// Migrates the engine's existing SQLite schema. @@ -307,6 +318,21 @@ impl Engine { Ok(()) } + /// Runs the existing schema-registry recovery for language-neutral metadata. + pub async fn reconcile_schema( + &self, + target: &SchemaTarget, + ) -> Result { + Ok(self.reconciler.reconcile_target(target).await?) + } + + /// Records a schema version after its recovery has durably completed. + pub async fn record_schema(&self, target: &SchemaTarget) -> Result<(), EngineError> { + self.reconciler.record_target(target).await?; + + Ok(()) + } + pub(crate) fn pool(&self) -> &SqlitePool { &self.pool } @@ -1237,6 +1263,7 @@ impl From for PersistenceError { Json(source) => Self::DeserializationError(Box::new(source)), Integer(source) => Self::UnknownError(Box::new(source)), JobFlush(source) => Self::UnknownError(Box::new(source)), + Schema(source) => Self::UnknownError(Box::new(source)), } } } @@ -1449,6 +1476,7 @@ mod tests { use tokio::sync::Notify; use super::*; + use crate::CompactionPolicy; use crate::job::{ JobId, JobKind, JobStoreError, WorkerId, enqueued_event, pending_seed_payload, plan_claim, }; @@ -1578,6 +1606,79 @@ mod tests { assert_eq!((observed, limit), (302, 64)); } + #[tokio::test] + async fn schema_reconciliation_uses_the_existing_event_sourced_registry() { + let engine = Engine::new(create_test_pool().await.unwrap()); + let target = SchemaTarget::new("engine-schema-test", 1, CompactionPolicy::Retain); + + assert_eq!( + engine.reconcile_schema(&target).await.unwrap(), + SchemaReconciliation::Changed + ); + assert_eq!( + engine.reconcile_schema(&target).await.unwrap(), + SchemaReconciliation::Changed + ); + + engine.record_schema(&target).await.unwrap(); + + assert_eq!( + engine.reconcile_schema(&target).await.unwrap(), + SchemaReconciliation::Unchanged + ); + } + + #[tokio::test] + async fn schema_version_change_clears_derived_snapshots() { + let engine = Engine::new(create_test_pool().await.unwrap()); + let stream = StreamIdentity::new("engine-schema-snapshot", "one"); + let version_one = SchemaTarget::new("engine-schema-snapshot", 1, CompactionPolicy::Retain); + assert_eq!( + engine.reconcile_schema(&version_one).await.unwrap(), + SchemaReconciliation::Changed + ); + engine.record_schema(&version_one).await.unwrap(); + persist_test_snapshot(&engine, &stream).await; + + let version_two = SchemaTarget::new("engine-schema-snapshot", 2, CompactionPolicy::Retain); + let outcome = engine.reconcile_schema(&version_two).await.unwrap(); + + assert_eq!(outcome, SchemaReconciliation::Changed); + assert_eq!(engine.load_snapshot(&stream).await.unwrap(), None); + } + + #[tokio::test] + async fn schema_version_change_preserves_compacted_snapshots_by_refusing_recovery() { + let engine = Engine::new(create_test_pool().await.unwrap()); + let stream = StreamIdentity::new("engine-schema-compact", "one"); + let version_one = SchemaTarget::new("engine-schema-compact", 1, CompactionPolicy::Retain); + assert_eq!( + engine.reconcile_schema(&version_one).await.unwrap(), + SchemaReconciliation::Changed + ); + engine.record_schema(&version_one).await.unwrap(); + persist_test_snapshot(&engine, &stream).await; + let preserved = engine.load_snapshot(&stream).await.unwrap().unwrap(); + let version_two = SchemaTarget::new( + "engine-schema-compact", + 2, + CompactionPolicy::CompactAfterSnapshot, + ); + + let result = engine.reconcile_schema(&version_two).await; + + assert!(matches!( + result, + Err(EngineError::Schema( + ReconcileError::CompactedSnapshotClear { .. } + )) + )); + assert_eq!( + engine.load_snapshot(&stream).await.unwrap(), + Some(preserved) + ); + } + #[tokio::test] async fn commit_and_load_use_the_existing_serialized_event_contract() { let engine = Engine::new(create_test_pool().await.unwrap()); @@ -2123,4 +2224,22 @@ mod tests { assert!(matches!(result, Err(EngineError::EmptySnapshotUpdate))); assert_eq!(engine.load_snapshot(&stream).await.unwrap(), None); } + + async fn persist_test_snapshot(engine: &Engine, stream: &StreamIdentity) { + let event = serialized_event(stream, 1); + engine + .commit(CommitRequest::new( + stream.clone(), + std::slice::from_ref(&event), + )) + .await + .unwrap(); + engine + .store_snapshot( + stream, + SnapshotWrite::new(serde_json::json!({ "value": 42 }), 1), + ) + .await + .unwrap(); + } } diff --git a/crates/event-sorcery/src/lib.rs b/crates/event-sorcery/src/lib.rs index 654b31a..a6b64ce 100644 --- a/crates/event-sorcery/src/lib.rs +++ b/crates/event-sorcery/src/lib.rs @@ -146,7 +146,9 @@ pub use reactor::{ IdempotentReactor, RETRY_BASE_DELAY_MS, RETRY_MAX_ATTEMPTS, RETRY_MAX_DELAY_MS, Reactor, RetryOnBusy, is_retryable_sqlite_busy, retry_with_backoff, }; -pub use schema_registry::{ReconcileError, Reconciler, SchemaReconciliation, SchemaRegistry}; +pub use schema_registry::{ + ReconcileError, Reconciler, SchemaReconciliation, SchemaRegistry, SchemaTarget, +}; pub use sqlite_event_repository::SqliteEventRepository; #[cfg(any(test, feature = "test-support"))] pub use testing::{ diff --git a/crates/event-sorcery/src/schema_registry.rs b/crates/event-sorcery/src/schema_registry.rs index 02cad9c..ce4d1df 100644 --- a/crates/event-sorcery/src/schema_registry.rs +++ b/crates/event-sorcery/src/schema_registry.rs @@ -37,6 +37,29 @@ use crate::{DomainEvent, Effect, EventSourced, Nil, uneventful}; /// Singleton aggregate ID for the schema registry. const REGISTRY_ID: &str = "schema"; +/// Language-neutral aggregate schema metadata used during startup recovery. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SchemaTarget { + aggregate_type: String, + version: u64, + compaction: CompactionPolicy, +} + +impl SchemaTarget { + #[must_use] + pub fn new( + aggregate_type: impl Into, + version: u64, + compaction: CompactionPolicy, + ) -> Self { + Self { + aggregate_type: aggregate_type.into(), + version, + compaction, + } + } +} + /// Tracks schema versions for all aggregates. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct SchemaRegistry { @@ -153,13 +176,89 @@ impl Reconciler { Self { cqrs, pool } } + /// Reconciles language-neutral schema metadata through the same registry + /// and snapshot recovery used by the native typed API. + pub async fn reconcile_target( + &self, + target: &SchemaTarget, + ) -> Result { + let stored_version = self + .load_registry() + .await? + .and_then(|registry| registry.version_of(&target.aggregate_type)); + + debug!( + target: "cqrs", + aggregate = target.aggregate_type, + ?stored_version, + "Loaded stored schema version" + ); + + if stored_version == Some(target.version) { + debug!( + target: "cqrs", + aggregate = target.aggregate_type, + version = target.version, + "Schema version unchanged" + ); + + return Ok(SchemaReconciliation::Unchanged); + } + + if stored_version.is_some() && target.compaction == CompactionPolicy::CompactAfterSnapshot { + return Err(ReconcileError::CompactedSnapshotClear { + aggregate: target.aggregate_type.clone(), + old_version: stored_version, + new_version: target.version, + }); + } + + sqlx::query!( + r#" + DELETE FROM snapshots + WHERE aggregate_type = ?1 + "#, + target.aggregate_type, + ) + .execute(&self.pool) + .await?; + + info!( + target: "cqrs", + aggregate = target.aggregate_type, + old_version = ?stored_version, + new_version = target.version, + "Cleared stale snapshots for schema version change" + ); + + Ok(SchemaReconciliation::Changed) + } + + /// Records a schema version only after its recovery has durably completed. + pub async fn record_target(&self, target: &SchemaTarget) -> Result<(), ReconcileError> { + self.cqrs + .execute( + REGISTRY_ID, + SchemaRegistryCommand::Register { + name: target.aggregate_type.clone(), + version: target.version, + }, + ) + .await?; + + Ok(()) + } + /// Rebuilds SchemaRegistry state from the full event log. async fn load_registry(&self) -> Result, ReconcileError> { - let payloads: Vec = sqlx::query_scalar( - "SELECT payload FROM events \ - WHERE aggregate_type = 'SchemaRegistry' \ - AND aggregate_id = 'schema' \ - ORDER BY sequence", + let payloads = sqlx::query_scalar!( + r#" + SELECT payload AS "payload: String" + FROM events + WHERE aggregate_type = 'SchemaRegistry' + AND aggregate_id = 'schema' + ORDER BY sequence + "#, ) .fetch_all(&self.pool) .await?; @@ -196,58 +295,12 @@ impl Reconciler { pub async fn reconcile( &self, ) -> Result { - let name = Entity::AGGREGATE_TYPE; - let current_version = Entity::SCHEMA_VERSION; - - let stored_version = self - .load_registry() - .await? - .and_then(|registry| registry.version_of(name)); - - debug!(target: "cqrs", aggregate = %name, ?stored_version, "Loaded stored schema version"); - - let needs_clear = stored_version != Some(current_version); - - if needs_clear { - // Compactable aggregates may have deleted their pre-snapshot - // events. Clearing their snapshots would permanently lose - // that state with no way to rebuild. Refuse and require an - // explicit migration instead. - // - // Skip the guard on first registration (stored_version = None): - // there are no snapshots to clear and no compacted events yet. - if stored_version.is_some() - && matches!( - Entity::COMPACTION_POLICY, - CompactionPolicy::CompactAfterSnapshot - ) - { - return Err(ReconcileError::CompactedSnapshotClear { - aggregate: name.to_string(), - old_version: stored_version, - new_version: current_version, - }); - } - - sqlx::query("DELETE FROM snapshots WHERE aggregate_type = ?") - .bind(name) - .execute(&self.pool) - .await?; - - info!( - target: "cqrs", - aggregate = name, - old_version = ?stored_version, - new_version = current_version, - "Cleared stale snapshots for schema version change" - ); - - Ok(SchemaReconciliation::Changed) - } else { - debug!(target: "cqrs", aggregate = %name, version = current_version, "Schema version unchanged"); - - Ok(SchemaReconciliation::Unchanged) - } + self.reconcile_target(&SchemaTarget::new( + Entity::AGGREGATE_TYPE, + Entity::SCHEMA_VERSION, + Entity::COMPACTION_POLICY, + )) + .await } /// Records that `Entity`'s current [`SCHEMA_VERSION`] has been fully @@ -257,17 +310,12 @@ impl Reconciler { /// /// [`SCHEMA_VERSION`]: crate::EventSourced::SCHEMA_VERSION pub async fn record_version(&self) -> Result<(), ReconcileError> { - self.cqrs - .execute( - REGISTRY_ID, - SchemaRegistryCommand::Register { - name: Entity::AGGREGATE_TYPE.to_string(), - version: Entity::SCHEMA_VERSION, - }, - ) - .await?; - - Ok(()) + self.record_target(&SchemaTarget::new( + Entity::AGGREGATE_TYPE, + Entity::SCHEMA_VERSION, + Entity::COMPACTION_POLICY, + )) + .await } }