fix: skip processing of receipts when there's no receipt for a version

Author: TypeLevelConsoli

Cargo.lock

@@ -63,6 +63,7 @@ pin-project = "1.1.7"
 tonic.workspace = true
 itertools = "0.14.0"
 
+
 [dev-dependencies]
 hex-literal = "0.4.1"
 test-assets = { path = "../test-assets" }
@@ -74,6 +75,7 @@ tokio-test = "0.4.4"
 wiremock.workspace = true
 insta = "1.41.1"
 test-log.workspace = true
+futures = "0.3.31"
 
 [build-dependencies]
 build-info-build = { version = "0.0.40", default-features = false }

crates/service/Cargo.toml

@@ -27,25 +27,50 @@ enum ProcessReceiptError {
     Both(anyhow::Error, anyhow::Error),
 }
 
+/// Indicates which versions of Receipts where processed
+/// It's intended to be used for migration tests
+#[derive(Debug, PartialEq, Eq)]
+pub enum Processed {
+    V1,
+    V2,
+    All,
+    None,
+}
+
 impl InnerContext {
     async fn process_db_receipts(
         &self,
         buffer: Vec<DatabaseReceipt>,
-    ) -> Result<(), ProcessReceiptError> {
+    ) -> Result<Processed, ProcessReceiptError> {
         let (v1_receipts, v2_receipts): (Vec<_>, Vec<_>) =
             buffer.into_iter().partition_map(|r| match r {
                 DatabaseReceipt::V1(db_receipt_v1) => Either::Left(db_receipt_v1),
                 DatabaseReceipt::V2(db_receipt_v2) => Either::Right(db_receipt_v2),
             });
-        let (insert_v1, insert_v2) = tokio::join!(
-            self.store_receipts_v1(v1_receipts),
-            self.store_receipts_v2(v2_receipts)
-        );
+
+        let (insert_v1, insert_v2) = match (v1_receipts.is_empty(), v2_receipts.is_empty()) {
+            (true, true) => (None, None),
+            (false, true) => (Some(self.store_receipts_v1(v1_receipts).await), None),
+            (true, false) => (None, Some(self.store_receipts_v2(v2_receipts).await)),
+            (false, false) => {
+                let (v1, v2) = tokio::join!(
+                    self.store_receipts_v1(v1_receipts),
+                    self.store_receipts_v2(v2_receipts),
+                );
+                (Some(v1), Some(v2))
+            }
+        };
+
         match (insert_v1, insert_v2) {
-            (Err(e1), Err(e2)) => Err(ProcessReceiptError::Both(e1.into(), e2.into())),
-            (Err(e1), _) => Err(ProcessReceiptError::V1(e1.into())),
-            (_, Err(e2)) => Err(ProcessReceiptError::V2(e2.into())),
-            _ => Ok(()),
+            (Some(Err(e1)), Some(Err(e2))) => Err(ProcessReceiptError::Both(e1.into(), e2.into())),
+            (Some(Err(e1)), _) => Err(ProcessReceiptError::V1(e1.into())),
+            (_, Some(Err(e2))) => Err(ProcessReceiptError::V2(e2.into())),
+
+            // only useful for testing
+            (Some(Ok(_)), None) => Ok(Processed::V1),
+            (None, Some(Ok(_))) => Ok(Processed::V2),
+            (Some(Ok(_)), Some(Ok(_))) => Ok(Processed::All),
+            (None, None) => Ok(Processed::None),
         }
     }
 
@@ -305,3 +330,176 @@ impl DbReceiptV2 {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::{path::PathBuf, sync::LazyLock};
+
+    use futures::future::BoxFuture;
+    use sqlx::{
+        migrate::{MigrationSource, Migrator},
+        PgPool,
+    };
+    use test_assets::{
+        create_signed_receipt, create_signed_receipt_v2, SignedReceiptRequest, INDEXER_ALLOCATIONS,
+        TAP_EIP712_DOMAIN,
+    };
+
+    use crate::tap::{
+        receipt_store::{
+            DatabaseReceipt, DbReceiptV1, DbReceiptV2, InnerContext, ProcessReceiptError, Processed,
+        },
+        AdapterError,
+    };
+
+    async fn create_v1() -> DatabaseReceipt {
+        let alloc = INDEXER_ALLOCATIONS.values().next().unwrap().clone();
+        let v1 = create_signed_receipt(
+            SignedReceiptRequest::builder()
+                .allocation_id(alloc.id)
+                .value(100)
+                .build(),
+        )
+        .await;
+        let v1 = DatabaseReceipt::V1(DbReceiptV1::from_receipt(&v1, &TAP_EIP712_DOMAIN).unwrap());
+        v1
+    }
+
+    async fn create_v2() -> DatabaseReceipt {
+        let v2 = create_signed_receipt_v2().call().await;
+        let v2 = DatabaseReceipt::V2(DbReceiptV2::from_receipt(&v2, &TAP_EIP712_DOMAIN).unwrap());
+        v2
+    }
+
+    mod when_all_migrations_are_run {
+        use super::*;
+
+        #[rstest::rstest]
+        #[case(Processed::None, async { vec![] })]
+        #[case(Processed::V1, async { vec![create_v1().await] })]
+        #[case(Processed::V2, async { vec![create_v2().await] })]
+        #[case(Processed::All, async { vec![create_v2().await, create_v1().await] })]
+        #[sqlx::test(migrations = "../../migrations")]
+        async fn v1_and_v2_are_processed_successfully(
+            #[ignore] pgpool: PgPool,
+            #[case] expected: Processed,
+            #[future(awt)]
+            #[case]
+            receipts: Vec<DatabaseReceipt>,
+        ) {
+            let context = InnerContext { pgpool };
+
+            let res = context.process_db_receipts(receipts).await.unwrap();
+
+            assert_eq!(res, expected);
+        }
+    }
+
+    mod when_horizon_migrations_are_ignored {
+        use super::*;
+
+        #[sqlx::test(migrator = "WITHOUT_HORIZON_MIGRATIONS")]
+        async fn test_empty_receipts_are_processed_successfully(pgpool: PgPool) {
+            let context = InnerContext { pgpool };
+
+            let res = context.process_db_receipts(vec![]).await.unwrap();
+
+            assert_eq!(res, Processed::None);
+        }
+
+        #[sqlx::test(migrator = "WITHOUT_HORIZON_MIGRATIONS")]
+        async fn test_v1_receipts_are_processed_successfully(pgpool: PgPool) {
+            let context = InnerContext { pgpool };
+
+            let v1 = create_v1().await;
+            let receipts = vec![v1];
+
+            let res = context.process_db_receipts(receipts).await.unwrap();
+
+            assert_eq!(res, Processed::V1);
+        }
+
+        #[rstest::rstest]
+        #[case(async { vec![create_v2().await] })]
+        #[case(async { vec![create_v2().await, create_v1().await] })]
+        #[sqlx::test(migrator = "WITHOUT_HORIZON_MIGRATIONS")]
+        async fn test_cases_with_v2_receipts_fails_to_process(
+            #[ignore] pgpool: PgPool,
+            #[future(awt)]
+            #[case]
+            receipts: Vec<DatabaseReceipt>,
+        ) {
+            let context = InnerContext { pgpool };
+
+            let error = context.process_db_receipts(receipts).await.unwrap_err();
+
+            let ProcessReceiptError::V2(error) = error else {
+                panic!()
+            };
+            let d = error.downcast_ref::<AdapterError>().unwrap().to_string();
+
+            assert_eq!(
+                d,
+                "error returned from database: relation \"tap_horizon_receipts\" does not exist"
+            );
+        }
+
+        pub static WITHOUT_HORIZON_MIGRATIONS: LazyLock<Migrator> = LazyLock::new(create_migrator);
+
+        pub fn create_migrator() -> Migrator {
+            futures::executor::block_on(Migrator::new(MigrationRunner::new(
+                "../../migrations",
+                ["horizon"],
+            )))
+            .unwrap()
+        }
+
+        #[derive(Debug)]
+        pub struct MigrationRunner {
+            migration_path: PathBuf,
+            ignored_migrations: Vec<String>,
+        }
+
+        impl MigrationRunner {
+            /// Construct a new MigrationRunner that does not apply the given migrations.
+            ///
+            /// `ignored_migrations` is any iterable of strings that describes which
+            /// migrations to be ignored.
+            pub fn new<I>(path: impl Into<PathBuf>, ignored_migrations: I) -> Self
+            where
+                I: IntoIterator,
+                I::Item: Into<String>,
+            {
+                Self {
+                    migration_path: path.into(),
+                    ignored_migrations: ignored_migrations.into_iter().map(Into::into).collect(),
+                }
+            }
+        }
+
+        impl MigrationSource<'static> for MigrationRunner {
+            fn resolve(
+                self,
+            ) -> BoxFuture<'static, Result<Vec<sqlx::migrate::Migration>, sqlx::error::BoxDynError>>
+            {
+                Box::pin(async move {
+                    let canonical = self.migration_path.canonicalize()?;
+                    let migrations_with_paths =
+                        sqlx::migrate::resolve_blocking(&canonical).unwrap();
+
+                    let migrations_with_paths = migrations_with_paths
+                        .into_iter()
+                        .filter(|(_, p)| {
+                            let path = p.to_str().unwrap();
+                            self.ignored_migrations
+                                .iter()
+                                .any(|ignored| !path.contains(ignored))
+                        })
+                        .collect::<Vec<_>>();
+
+                    Ok(migrations_with_paths.into_iter().map(|(m, _p)| m).collect())
+                })
+            }
+        }
+    }
+}