test: pin atomic event and job persistence

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;

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<Value, serde_json::Error> {
+            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::<TestAggregate>(&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.