Fix out of order stream (#240)

* fix: out of order stream

Co-authored-by: Luca Iachini <luca.iachini89@gmail.com>

* bump disintegrate-postgres version

Co-authored-by: Luca Iachini <luca.iachini89@gmail.com>

* clippy fix

---------

Co-authored-by: Valerio Iachini <valerio.iachini@gmail.com>

Author: luca-iachini

Cargo.lock

@@ -77,15 +77,15 @@ To add Disintegrate to your project, follow these steps:
     ```toml
     [dependencies]
     disintegrate = "4.0.0"
-    disintegrate-postgres = "4.0.0"
+    disintegrate-postgres = "4.0.1"
     ```
 
     * Disintegrate provides several features that you can enable based on your project requirements. You can include them in your `Cargo.toml` file as follows:
 
     ```toml
     [dependencies]
     disintegrate = { version = "4.0.0", features = ["macros", "serde-prost"] }
-    disintegrate-postgres = { version = "4.0.0", features = ["listener"] }
+    disintegrate-postgres = { version = "4.0.1", features = ["listener"] }
     ```
 
     * The macros feature enables the use of derive macros to simplify events implementations.
@@ -97,7 +97,7 @@ To add Disintegrate to your project, follow these steps:
         * To enable Prost serialization, use the `serde-prost` feature: `features = ["serde-prost"]`.
         * To enable Protocol Buffers serialization, use the `serde-protobuf` feature: `features = ["serde-protobuf"]`.
 
-    * If you're using the PostgreSQL event store backend and want to use the listener mechanism, you can enable the `listener` feature: `disintegrate-postgres = {version = "4.0.0", features = ["listener"]}`.
+    * If you're using the PostgreSQL event store backend and want to use the listener mechanism, you can enable the `listener` feature: `disintegrate-postgres = {version = "4.0.1", features = ["listener"]}`.
 
 2. Define the list of events in your application. You can use the Event Storming technique to identify the events that occur in your system. Here's an example of defining events using Disintegrate:
 

README.md

@@ -1,7 +1,7 @@
 [package]
 name = "disintegrate-postgres"
 description = "Disintegrate PostgresDB implementation. Not for direct use. Refer to the `disintegrate` crate for details."
-version = "4.0.0"
+version = "4.0.1"
 license.workspace = true
 edition.workspace = true
 authors.workspace = true

disintegrate-postgres/Cargo.toml

@@ -84,17 +84,78 @@ where
     S: Serde<E> + Send + Sync,
     E: Event + Send + Sync,
 {
+    /// Appends events within an existing transaction without committing.
+    ///
+    /// This allows callers to control when the transaction is committed.
+    /// Sets SERIALIZABLE isolation, calls `event_store_begin_epoch()` to
+    /// signal an in-flight write, performs the concurrency check, and inserts events.
+    pub(crate) async fn append_in_tx<QE>(
+        &self,
+        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+        events: Vec<E>,
+        query: StreamQuery<PgEventId, QE>,
+        version: PgEventId,
+    ) -> Result<Vec<PersistedEvent<PgEventId, E>>, Error>
+    where
+        E: Clone,
+        QE: Event + Clone + Send + Sync,
+    {
+        sqlx::query("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
+            .execute(&mut **tx)
+            .await?;
+
+        sqlx::query("SELECT event_store_begin_epoch()")
+            .execute(&mut **tx)
+            .await?;
+
+        if sqlx::query_scalar(&format!(
+            "SELECT EXISTS (SELECT 1 FROM event WHERE {})",
+            CriteriaBuilder::new(&query.change_origin(version)).build()
+        ))
+        .fetch_one(&mut **tx)
+        .await?
+        {
+            return Err(Error::Concurrency);
+        }
+
+        let mut insert = InsertEventsBuilder::new(&events, &self.serde);
+        let event_ids: Vec<PgEventId> = insert
+            .build()
+            .fetch_all(&mut **tx)
+            .await?
+            .into_iter()
+            .map(|r| r.get(0))
+            .collect();
+
+        let persisted_events = event_ids
+            .iter()
+            .zip(events)
+            .map(|(event_id, event)| PersistedEvent::new(*event_id, event))
+            .collect::<Vec<_>>();
+
+        Ok(persisted_events)
+    }
+
     /// Streams events based on the provided query and executor.
     ///
+    /// Events are returned up to the current epoch, which is the minimum between
+    /// the lowest in-flight writer's sequence position and the highest committed
+    /// event ID. This guarantees that readers never observe uncommitted events
+    /// and never skip events when concurrent writers commit out of sequence order.
+    ///
+    /// The stream yields `StreamItem::Event` for each matching event and ends
+    /// with `StreamItem::End(epoch_id)` indicating the epoch up to which events
+    /// were read.
+    ///
     /// # Arguments
     ///
     /// * `executor` - The sqlx executor to use for database operations.
     /// * `query` - The stream query specifying the criteria for filtering events.
     ///
     /// # Returns
     ///
-    /// A `Result` containing a boxed stream of `PersistedEvent` that matches the query criteria,
-    /// or an error of type `Error`.
+    /// A boxed stream of `StreamItem` that contains matching events up to the
+    /// current epoch, or an error of type `Error`.
     pub(crate) fn stream_with<'a, QE, EX>(
         &'a self,
         executor: EX,
@@ -107,7 +168,7 @@ where
     {
         let sql = format!(
             r#"SELECT event.event_id, event.payload, epoch.__epoch_id
-            FROM (SELECT COALESCE(MAX(event_id),0) AS __epoch_id FROM event) AS epoch
+            FROM (values (event_store_current_epoch())) AS epoch(__epoch_id)
             LEFT JOIN event ON event.event_id <= epoch.__epoch_id AND ({criteria})
             ORDER BY event_id ASC"#,
             criteria = CriteriaBuilder::new(query).build()
@@ -203,34 +264,8 @@ where
         QE: Event + Clone + Send + Sync,
     {
         let mut tx = self.pool.begin().await?;
-        sqlx::query("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
-            .execute(&mut *tx)
-            .await?;
 
-        if sqlx::query_scalar(&format!(
-            "SELECT EXISTS (SELECT 1 FROM event WHERE {})",
-            CriteriaBuilder::new(&query.change_origin(version)).build()
-        ))
-        .fetch_one(&mut *tx)
-        .await?
-        {
-            return Err(Error::Concurrency);
-        }
-
-        let mut insert = InsertEventsBuilder::new(&events, &self.serde);
-        let event_ids: Vec<PgEventId> = insert
-            .build()
-            .fetch_all(&mut *tx)
-            .await?
-            .into_iter()
-            .map(|r| r.get(0))
-            .collect();
-
-        let persisted_events = event_ids
-            .iter()
-            .zip(events)
-            .map(|(event_id, event)| PersistedEvent::new(*event_id, event))
-            .collect::<Vec<_>>();
+        let persisted_events = self.append_in_tx(&mut tx, events, query, version).await?;
 
         tx.commit().await.map_err(map_concurrency_err)?;
 
@@ -257,6 +292,10 @@ where
     {
         let mut tx = self.pool.begin().await?;
 
+        sqlx::query("SELECT event_store_begin_epoch()")
+            .execute(&mut *tx)
+            .await?;
+
         let mut insert = InsertEventsBuilder::new(&events, &self.serde);
         let event_ids: Vec<PgEventId> = insert
             .build()

disintegrate-postgres/src/event_store.rs

@@ -0,0 +1,14 @@
+CREATE OR REPLACE FUNCTION event_store_begin_epoch()
+RETURNS void AS $$
+DECLARE
+    id BIGINT;
+    db_id INT;
+BEGIN
+    SELECT last_value INTO id FROM seq_event_event_id;
+    SELECT oid INTO db_id FROM pg_database WHERE datname = current_database();
+
+    PERFORM pg_try_advisory_xact_lock_shared(db_id, 0);
+    PERFORM pg_try_advisory_xact_lock_shared(1, (id & 0xFFFFFFFF)::bit(32)::integer);
+    PERFORM pg_try_advisory_xact_lock_shared(2, (id >> 32)::bit(32)::integer);
+END;
+$$ LANGUAGE plpgsql;

disintegrate-postgres/src/event_store/sql/fn_event_store_begin_epoch.sql

@@ -0,0 +1,24 @@
+CREATE OR REPLACE FUNCTION event_store_current_epoch()
+RETURNS BIGINT AS $$
+DECLARE
+    persisted_event_id BIGINT;
+    pending_event_id BIGINT;
+    db_id INT;
+BEGIN
+    SELECT COALESCE(MAX(event_id), 0) INTO persisted_event_id FROM event;
+    SELECT oid INTO db_id FROM pg_database WHERE datname = current_database();
+
+    SELECT MIN((l3.objid::bigint << 32) + l2.objid::bigint)
+    INTO pending_event_id
+    FROM pg_locks l1
+    INNER JOIN pg_locks l2 ON l1.pid = l2.pid
+    INNER JOIN pg_locks l3 ON l1.pid = l3.pid
+    WHERE
+        l1.classid = db_id
+        AND l2.classid = 1
+        AND l3.classid = 2
+        AND l1.locktype = 'advisory';
+
+    RETURN LEAST(COALESCE(pending_event_id, persisted_event_id), persisted_event_id);
+END;
+$$ LANGUAGE plpgsql;