feat(fortuna): use requestV2 event, parse revealed event from tx logs, add gas and callback fields to history API

Author: tejasbadadare

apps/fortuna/migrations/20250801232728_add_callback_fields.down.sql

@@ -0,0 +1,4 @@
+-- Remove callback_failed, callback_return_value, and callback_gas_used from the requests table.
+ALTER TABLE request DROP COLUMN callback_failed;
+ALTER TABLE request DROP COLUMN callback_return_value;
+ALTER TABLE request DROP COLUMN callback_gas_used;

apps/fortuna/migrations/20250801232728_add_callback_fields.up.sql

@@ -0,0 +1,4 @@
+-- Add callback_failed, callback_return_value, and callback_gas_used to the requests table.
+ALTER TABLE request ADD COLUMN callback_failed INTEGER;
+ALTER TABLE request ADD COLUMN callback_return_value VARCHAR;
+ALTER TABLE request ADD COLUMN callback_gas_used VARCHAR(100);

apps/fortuna/src/eth_utils/utils.rs

@@ -1,12 +1,13 @@
 use {
     crate::{
-        chain::ethereum::InstrumentedSignablePythContract, eth_utils::nonce_manager::NonceManaged,
+        chain::ethereum::{InstrumentedSignablePythContract, PythRandomEvents, Revealed2Filter},
+        eth_utils::nonce_manager::NonceManaged,
     },
     anyhow::{anyhow, Result},
     backoff::ExponentialBackoff,
     ethabi::ethereum_types::U64,
     ethers::{
-        contract::{ContractCall, ContractError},
+        contract::{ContractCall, ContractError, EthLogDecode},
         middleware::Middleware,
         providers::{MiddlewareError, ProviderError},
         signers::Signer,
@@ -30,6 +31,7 @@ pub struct SubmitTxResult {
     pub fee_multiplier: u64,
     pub duration: Duration,
     pub receipt: TransactionReceipt,
+    pub revealed_event: Revealed2Filter,
 }
 
 #[derive(Clone, Debug)]
@@ -186,11 +188,26 @@ pub async fn submit_tx_with_backoff<T: Middleware + NonceManaged + 'static>(
     let duration = start_time.elapsed();
     let num_retries = num_retries.load(std::sync::atomic::Ordering::Relaxed);
 
+    let revealed_event: Revealed2Filter = {
+        let mut found_event = None;
+        for log in &success.logs {
+            if let Ok(PythRandomEvents::Revealed2Filter(decoded_event)) =
+                PythRandomEvents::decode_log(&log.clone().into())
+            {
+                found_event = Some(decoded_event);
+                break;
+            }
+        }
+        // A successful reveal will always emit a Revealed v2 event, so theoretically we should never get here.
+        found_event.ok_or_else(|| SubmitTxError::ReceiptError(call.tx.clone(), success.clone()))?
+    };
+
     Ok(SubmitTxResult {
         num_retries,
         fee_multiplier: escalation_policy.get_fee_multiplier_pct(num_retries),
         duration,
         receipt: success,
+        revealed_event,
     })
 }
 

apps/fortuna/src/history.rs

@@ -5,7 +5,7 @@ use {
     ethers::{
         core::utils::hex::ToHex,
         prelude::TxHash,
-        types::{Address, U256},
+        types::{Address, Bytes, U256},
         utils::keccak256,
     },
     serde::Serialize,
@@ -45,6 +45,17 @@ pub enum RequestEntryState {
         #[schema(example = "a905ab56567d31a7fda38ed819d97bc257f3ebe385fc5c72ce226d3bb855f0fe")]
         #[serde_as(as = "serde_with::hex::Hex")]
         combined_random_number: [u8; 32],
+        /// Whether the callback to the caller failed.
+        callback_failed: bool,
+        /// Return value from the callback. If the callback failed, this field contains
+        /// the error code and any additional returned data. Note that "" often indicates an out-of-gas error.
+        /// If the callback returns more than 256 bytes, only the first 256 bytes of the callback return value are included.
+        /// NOTE: This field is the raw bytes returned from the callback, not hex-decoded. The client should decode it as needed.
+        callback_return_value: Bytes,
+        /// How much gas the callback used.
+        #[schema(example = "567890", value_type = String)]
+        #[serde(with = "crate::serde::u32")]
+        callback_gas_used: u32,
     },
     Failed {
         reason: String,
@@ -120,6 +131,9 @@ struct RequestRow {
     provider_random_number: Option<String>,
     gas_used: Option<String>,
     info: Option<String>,
+    callback_failed: Option<i64>,
+    callback_return_value: Option<String>,
+    callback_gas_used: Option<String>,
 }
 
 impl TryFrom<RequestRow> for RequestStatus {
@@ -174,6 +188,18 @@ impl TryFrom<RequestRow> for RequestStatus {
                         &user_random_number,
                         &provider_random_number,
                     ),
+                    //  Sqlx::Any doesn't support boolean types, so we need to convert from integer
+                    //  https://github.com/launchbadge/sqlx/issues/2778
+                    callback_failed: row.callback_failed.unwrap_or(0) == 1,
+                    callback_return_value: row
+                        .callback_return_value
+                        .map(|s| s.parse::<Bytes>().unwrap_or_default())
+                        .unwrap_or_default(),
+                    callback_gas_used: row
+                        .callback_gas_used
+                        .unwrap_or_default()
+                        .parse::<u32>()
+                        .map_err(|_| anyhow::anyhow!("Failed to parse callback_gas_used"))?,
                 }
             }
             "Failed" => RequestEntryState::Failed {
@@ -313,18 +339,29 @@ impl History {
                 provider_random_number,
                 gas_used,
                 combined_random_number: _,
+                callback_failed,
+                callback_return_value,
+                callback_gas_used,
             } => {
                 let reveal_block_number = reveal_block_number as i64;
                 let reveal_tx_hash: String = reveal_tx_hash.encode_hex();
                 let provider_random_number: String = provider_random_number.encode_hex();
                 let gas_used: String = gas_used.to_string();
-                let result = sqlx::query("UPDATE request SET state = $1, last_updated_at = $2, reveal_block_number = $3, reveal_tx_hash = $4, provider_random_number = $5, gas_used = $6 WHERE network_id = $7 AND sequence = $8 AND provider = $9 AND request_tx_hash = $10")
+                //  Sqlx::Any doesn't support boolean types, so we need to convert to integer
+                //  https://github.com/launchbadge/sqlx/issues/2778
+                let callback_failed: i64 = if callback_failed { 1 } else { 0 };
+                let callback_return_value: String = callback_return_value.encode_hex();
+                let callback_gas_used: String = callback_gas_used.to_string();
+                let result = sqlx::query("UPDATE request SET state = $1, last_updated_at = $2, reveal_block_number = $3, reveal_tx_hash = $4, provider_random_number = $5, gas_used = $6, callback_failed = $7, callback_return_value = $8, callback_gas_used = $9 WHERE network_id = $10 AND sequence = $11 AND provider = $12 AND request_tx_hash = $13")
                     .bind("Completed")
                     .bind(new_status.last_updated_at.timestamp())
                     .bind(reveal_block_number)
                     .bind(reveal_tx_hash)
                     .bind(provider_random_number)
                     .bind(gas_used)
+                    .bind(callback_failed)
+                    .bind(callback_return_value)
+                    .bind(callback_gas_used)
                     .bind(network_id)
                     .bind(sequence)
                     .bind(provider.clone())
@@ -666,6 +703,9 @@ mod test {
                 &status.user_random_number,
                 &[40; 32],
             ),
+            callback_failed: false,
+            callback_return_value: Default::default(),
+            callback_gas_used: 100_000,
         };
         History::update_request_status(&history.pool, status.clone()).await;
 
@@ -877,6 +917,9 @@ mod test {
                 &status.user_random_number,
                 &[40; 32],
             ),
+            callback_failed: false,
+            callback_return_value: Default::default(),
+            callback_gas_used: 0,
         };
         History::update_request_status(&history.pool, status.clone()).await;
         let mut failed_status = status.clone();

apps/fortuna/src/keeper/process_event.rs

@@ -192,6 +192,9 @@ pub async fn process_event_with_backoff(
                     &event.user_random_number,
                     &provider_revelation,
                 ),
+                callback_failed: result.revealed_event.callback_failed,
+                callback_return_value: result.revealed_event.callback_return_value,
+                callback_gas_used: result.revealed_event.callback_gas_used,
             };
             history.add(&status);
             tracing::info!(

apps/fortuna/src/serde.rs

@@ -19,3 +19,23 @@ pub mod u256 {
         U256::from_dec_str(s.as_str()).map_err(|err| D::Error::custom(err.to_string()))
     }
 }
+
+pub mod u32 {
+    use serde::{de::Error, Deserialize, Deserializer, Serializer};
+
+    pub fn serialize<S>(n: &u32, s: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        s.serialize_str(&n.to_string())
+    }
+
+    pub fn deserialize<'de, D>(d: D) -> Result<u32, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let s: String = Deserialize::deserialize(d)?;
+        s.parse::<u32>()
+            .map_err(|err| D::Error::custom(err.to_string()))
+    }
+}