graph: Detect unique constraint violations and mark as deterministic

Author: lutter

graph/src/components/store/err.rs

@@ -70,6 +70,8 @@ pub enum StoreError {
     WriteFailure(String, BlockNumber, String, String),
     #[error("database query timed out")]
     StatementTimeout,
+    #[error("database constraint violated: {0}")]
+    ConstraintViolation(String),
 }
 
 // Convenience to report an internal error
@@ -127,6 +129,7 @@ impl Clone for StoreError {
                 Self::WriteFailure(arg0.clone(), arg1.clone(), arg2.clone(), arg3.clone())
             }
             Self::StatementTimeout => Self::StatementTimeout,
+            Self::ConstraintViolation(arg0) => Self::ConstraintViolation(arg0.clone()),
         }
     }
 }
@@ -135,6 +138,7 @@ impl StoreError {
     pub fn from_diesel_error(e: &DieselError) -> Option<Self> {
         const CONN_CLOSE: &str = "server closed the connection unexpectedly";
         const STMT_TIMEOUT: &str = "canceling statement due to statement timeout";
+        const UNIQUE_CONSTR: &str = "duplicate key value violates unique constraint";
         let DieselError::DatabaseError(_, info) = e else {
             return None;
         };
@@ -146,6 +150,12 @@ impl StoreError {
             Some(StoreError::DatabaseUnavailable)
         } else if info.message().contains(STMT_TIMEOUT) {
             Some(StoreError::StatementTimeout)
+        } else if info.message().contains(UNIQUE_CONSTR) {
+            let msg = match info.details() {
+                Some(details) => format!("{}: {}", info.message(), details.replace('\n', " ")),
+                None => info.message().to_string(),
+            };
+            Some(StoreError::ConstraintViolation(msg))
         } else {
             None
         }
@@ -174,7 +184,8 @@ impl StoreError {
             | UnknownTable(_)
             | UnknownAttribute(_, _)
             | InvalidIdentifier(_)
-            | UnsupportedFilter(_, _) => true,
+            | UnsupportedFilter(_, _)
+            | ConstraintViolation(_) => true,
 
             // non-deterministic errors
             Unknown(_)

store/test-store/tests/postgres/writable.rs

@@ -449,6 +449,7 @@ fn read_range_pool_created_test() {
         let pool_created_type = TEST_SUBGRAPH_SCHEMA.entity_type("PoolCreated").unwrap();
         let entity_types = vec![pool_created_type.clone()];
 
+        let mut last_op: Option<EntityOperation> = None;
         for count in (1..=2).map(|x| x as i64) {
             let id = if count == 1 {
                 "0xff80818283848586"
@@ -478,6 +479,7 @@ fn read_range_pool_created_test() {
                 data,
             };
 
+            last_op = Some(op.clone());
             transact_entity_operations(
                 &subgraph_store,
                 &deployment,
@@ -500,5 +502,21 @@ fn read_range_pool_created_test() {
             let a = result_entities[index as usize].clone();
             assert_eq!(a, format!("{:?}", en));
         }
+
+        // Make sure we get a constraint violation
+        let op = last_op.take().unwrap();
+
+        transact_entity_operations(&subgraph_store, &deployment, block_pointer(3), vec![op])
+            .await
+            .unwrap();
+        let res = writable.flush().await;
+        let exp = "duplicate key value violates unique constraint \"pool_created_pkey\": Key (vid)=(2) already exists.";
+        match res {
+            Ok(_) => panic!("Expected error, but got success"),
+            Err(StoreError::ConstraintViolation(msg)) => {
+                assert_eq!(msg, exp);
+            }
+            Err(e) => panic!("Expected constraint violation, but got {:?}", e),
+        }
     })
 }