diff --git a/.sqlx/query-19871c098856d59714f49325ce46ed9bbc172ba4c2784cca79a8cc363ce343fe.json b/.sqlx/query-19871c098856d59714f49325ce46ed9bbc172ba4c2784cca79a8cc363ce343fe.json new file mode 100644 index 0000000..9d4d7df --- /dev/null +++ b/.sqlx/query-19871c098856d59714f49325ce46ed9bbc172ba4c2784cca79a8cc363ce343fe.json @@ -0,0 +1,21 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT COUNT(*)\n FROM job_queue\n WHERE view_id = ?1\n ", + "describe": { + "columns": [ + { + "name": "COUNT(*)", + "ordinal": 0, + "type_info": "Integer", + "origin": "Expression" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "19871c098856d59714f49325ce46ed9bbc172ba4c2784cca79a8cc363ce343fe" +} diff --git a/.sqlx/query-24c9c228c55b06ef7f1d461b39cf5e7258af444bc8a3e591fdf7fa7bc923fe27.json b/.sqlx/query-24c9c228c55b06ef7f1d461b39cf5e7258af444bc8a3e591fdf7fa7bc923fe27.json new file mode 100644 index 0000000..32b8bbb --- /dev/null +++ b/.sqlx/query-24c9c228c55b06ef7f1d461b39cf5e7258af444bc8a3e591fdf7fa7bc923fe27.json @@ -0,0 +1,21 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT COUNT(*)\n FROM events\n WHERE aggregate_type = ?1 AND aggregate_id = ?2\n ", + "describe": { + "columns": [ + { + "name": "COUNT(*)", + "ordinal": 0, + "type_info": "Integer", + "origin": "Expression" + } + ], + "parameters": { + "Right": 2 + }, + "nullable": [ + false + ] + }, + "hash": "24c9c228c55b06ef7f1d461b39cf5e7258af444bc8a3e591fdf7fa7bc923fe27" +} diff --git a/.sqlx/query-8583cfb342c957b555202f46b51ece85ef7fb66072ce129cb3f2f65b12c6e0e3.json b/.sqlx/query-8583cfb342c957b555202f46b51ece85ef7fb66072ce129cb3f2f65b12c6e0e3.json new file mode 100644 index 0000000..b51e344 --- /dev/null +++ b/.sqlx/query-8583cfb342c957b555202f46b51ece85ef7fb66072ce129cb3f2f65b12c6e0e3.json @@ -0,0 +1,21 @@ +{ + "db_name": "SQLite", + "query": "\n SELECT COUNT(*)\n FROM events\n WHERE aggregate_type = 'job' AND aggregate_id = ?1\n ", + "describe": { + "columns": [ + { + "name": "COUNT(*)", + "ordinal": 0, + "type_info": "Integer", + "origin": "Expression" + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "8583cfb342c957b555202f46b51ece85ef7fb66072ce129cb3f2f65b12c6e0e3" +} diff --git a/crates/event-sorcery/src/job_backend.rs b/crates/event-sorcery/src/job_backend.rs index c70bea5..8dd64c4 100644 --- a/crates/event-sorcery/src/job_backend.rs +++ b/crates/event-sorcery/src/job_backend.rs @@ -1113,6 +1113,33 @@ mod tests { ); } + #[tokio::test] + async fn job_queue_rebuilds_from_the_job_event_stream() { + let pool = one_db_pool().await; + let runtime = JobRuntime::build(pool.clone()).await.unwrap(); + let job_id = runtime.enqueue(TestJob { n: 7 }).await.unwrap(); + + sqlx::query( + r" + DELETE FROM job_queue + WHERE view_id = ?1 + ", + ) + .bind(job_id.to_string()) + .execute(&pool) + .await + .unwrap(); + assert_eq!(queue_status(&pool, &job_id).await, None); + + let _rebuilt = JobRuntime::build(pool.clone()).await.unwrap(); + + assert_eq!(event_types(&pool, &job_id).await, ["JobEnqueued"]); + assert_eq!( + queue_status(&pool, &job_id).await.as_deref(), + Some("pending") + ); + } + #[tokio::test] async fn standalone_enqueue_with_delay_is_not_runnable_until_run_at() { let pool = one_db_pool().await; diff --git a/crates/event-sorcery/src/sqlite_event_repository.rs b/crates/event-sorcery/src/sqlite_event_repository.rs index b7f6a56..8b34a41 100644 --- a/crates/event-sorcery/src/sqlite_event_repository.rs +++ b/crates/event-sorcery/src/sqlite_event_repository.rs @@ -388,6 +388,12 @@ mod tests { use std::fmt::{self, Display}; use super::*; + use crate::SqliteBackend; + use crate::job::{ + EnqueueRequest, ErasedJob, JobId, JobKind, PendingPush, enqueued_event, + pending_seed_payload, + }; + use crate::job_store::EventBackend; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] struct TestAggregate { @@ -412,6 +418,18 @@ mod tests { #[derive(Debug)] struct TestError; + struct TestPendingJob; + + impl ErasedJob for TestPendingJob { + fn kind(&self) -> JobKind { + JobKind::new("test-pending") + } + + fn encode(&self) -> Result { + Ok(serde_json::json!({ "payload": "test" })) + } + } + impl Display for TestError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "test error") @@ -461,6 +479,101 @@ mod tests { .collect() } + async fn persist_with_pending_job( + pool: SqlitePool, + aggregate_id: &str, + job_id: JobId, + ) -> Result<(), PersistenceError> { + let repo = SqliteEventRepository::new(pool, CompactionPolicy::Retain); + crate::job::with_pending_scope(async move { + crate::job::buffer(PendingPush { + job_id, + job: Box::new(TestPendingJob), + delay: None, + }) + .unwrap(); + + repo.persist::(&covering_events(aggregate_id, 1), None) + .await + }) + .await + } + + async fn count_events(pool: &SqlitePool, aggregate_type: &str, aggregate_id: &str) -> i64 { + sqlx::query_scalar!( + r" + SELECT COUNT(*) + FROM events + WHERE aggregate_type = ?1 AND aggregate_id = ?2 + ", + aggregate_type, + aggregate_id, + ) + .fetch_one(pool) + .await + .unwrap() + } + + async fn count_queue_rows(pool: &SqlitePool, job_id: &str) -> i64 { + sqlx::query_scalar!( + r" + SELECT COUNT(*) + FROM job_queue + WHERE view_id = ?1 + ", + job_id, + ) + .fetch_one(pool) + .await + .unwrap() + } + + #[tokio::test] + async fn aggregate_event_job_event_and_queue_seed_commit_together() { + let pool = test_pool().await; + let job_id = JobId::new(); + + persist_with_pending_job(pool.clone(), "atomic-success", job_id) + .await + .unwrap(); + + let aggregate_events = count_events(&pool, TestAggregate::TYPE, "atomic-success").await; + let job_id = job_id.to_string(); + let job_events = count_events(&pool, "job", &job_id).await; + let queue_rows = count_queue_rows(&pool, &job_id).await; + + assert_eq!((aggregate_events, job_events, queue_rows), (1, 1, 1)); + } + + #[tokio::test] + async fn job_identity_conflict_rolls_back_the_aggregate_event() { + let pool = test_pool().await; + let job_id = JobId::new(); + let request = EnqueueRequest { + job_id, + kind: JobKind::new("existing-job"), + payload: serde_json::json!({ "payload": "existing" }), + run_at_ms: 0, + }; + SqliteBackend::new(pool.clone()) + .enqueue( + enqueued_event(&request).unwrap(), + pending_seed_payload(&request).unwrap(), + ) + .await + .unwrap(); + + let result = persist_with_pending_job(pool.clone(), "atomic-rollback", job_id).await; + + assert!(matches!(result, Err(PersistenceError::OptimisticLockError))); + let aggregate_events = count_events(&pool, TestAggregate::TYPE, "atomic-rollback").await; + let job_id = job_id.to_string(); + let existing_job_events = count_events(&pool, "job", &job_id).await; + + assert_eq!(aggregate_events, 0); + assert_eq!(existing_job_events, 1); + } + /// For a `Retain` entity, get_snapshot returns the incompatible snapshot /// unchanged (no shape check, no discard); cqrs-es then rebuilds from the /// always-complete event history. See the end-to-end rebuild test below.