chore: upgrade Pallas to v0.30.1 (#812)

Cargo.toml

@@ -23,7 +23,7 @@ mithril = ["mithril-client"]
 # kafka = auto feature flag
 
 [dependencies]
-pallas = { version = "0.27.0", features = ["hardano"] }
+pallas = { version = "0.30.1", features = ["hardano"] }
 # pallas = { path = "../pallas/pallas" }
 # pallas = { git = "https://github.com/txpipe/pallas" }
 
@@ -32,8 +32,6 @@ gasket-prometheus = { version = "^0.7" }
 # gasket = { path = "../../construkts/gasket-rs/gasket", features = ["derive"] }
 # gasket = { git = "https://github.com/construkts/gasket-rs.git", features = ["derive"] }
 
-utxorpc-spec = { version = "^0.5" }
-
 hex = "0.4.3"
 net2 = "0.2.37"
 bech32 = "0.9.1"
@@ -80,4 +78,4 @@ aws-sdk-lambda = { version = "^1.1", optional = true }
 extism = { version = "1.2.0", optional = true }
 mithril-client = { version = "^0.8", optional = true, features = ["fs"] }
 miette = { version = "7.2.0", features = ["fancy"] }
-itertools = "0.12.1"
+itertools = "0.12.1"

src/filters/legacy_v1/map.rs

@@ -408,7 +408,7 @@ impl EventWriter<'_> {
                 conway::RedeemerTag::Vote => "vote".to_string(),
                 conway::RedeemerTag::Propose => "propose".to_string(),
             },
-            ex_units_mem: redeemer.ex_units().mem,
+            ex_units_mem: redeemer.ex_units().mem as u32,
             ex_units_steps: redeemer.ex_units().steps,
             input_idx: redeemer.index(),
             plutus_data: redeemer.data().to_json(),

src/filters/parse_cbor.rs

@@ -11,12 +11,8 @@ use crate::framework::*;
 #[derive(Clone, Default)]
 struct NoOpContext;
 
-impl interop::Context for NoOpContext {
-    fn get_txo<'a>(
-        &self,
-        _tx_hash: pallas::crypto::hash::Hash<32>,
-        _txo_index: u32,
-    ) -> Option<trv::MultiEraOutput<'a>> {
+impl interop::LedgerContext for NoOpContext {
+    fn get_utxos(&self, _refs: &[interop::TxoRef]) -> Option<interop::UtxoMap> {
         None
     }
 }

src/filters/select/eval/mod.rs

@@ -1,10 +1,10 @@
 use std::{ops::Deref, str::FromStr};
 
-use serde::{Deserialize, Serialize};
-use tracing::warn;
-use utxorpc_spec::utxorpc::v1alpha::cardano::{
+use pallas::interop::utxorpc::spec::cardano::{
     Asset, AuxData, Metadata, Metadatum, Multiasset, TxInput, TxOutput,
 };
+use serde::{Deserialize, Serialize};
+use tracing::warn;
 
 use crate::framework::*;
 
@@ -129,6 +129,10 @@ pub trait PatternOf<S> {
         let outcomes = iter.map(|x| self.is_match(x));
         MatchOutcome::fold_any_of(outcomes)
     }
+
+    fn is_some_match(&self, subject: Option<S>) -> MatchOutcome {
+        self.is_any_match(subject.into_iter())
+    }
 }
 
 impl<S, P> PatternOf<S> for Option<P>
@@ -218,7 +222,7 @@ impl PatternOf<&[u8]> for TextPattern {
 impl PatternOf<&Metadatum> for TextPattern {
     fn is_match(&self, subject: &Metadatum) -> MatchOutcome {
         match subject.metadatum.as_ref() {
-            Some(utxorpc_spec::utxorpc::v1alpha::cardano::metadatum::Metadatum::Text(subject)) => {
+            Some(pallas::interop::utxorpc::spec::cardano::metadatum::Metadatum::Text(subject)) => {
                 self.is_match(subject.as_str())
             }
             _ => MatchOutcome::Negative,
@@ -289,7 +293,9 @@ impl PatternOf<&TxOutput> for OutputPattern {
 
         let c = MatchOutcome::fold_all_of(c);
 
-        let d = self.datum.is_match(subject.datum_hash.as_ref());
+        let d = self
+            .datum
+            .is_some_match(subject.datum.as_ref().map(|d| d.hash.as_ref()));
 
         MatchOutcome::fold_all_of([a, b, c, d].into_iter())
     }
@@ -330,7 +336,9 @@ impl PatternOf<&TxInput> for InputPattern {
 
         let c = MatchOutcome::fold_all_of(c);
 
-        let d = self.datum.is_match(as_output.datum_hash.as_ref());
+        let d = self
+            .datum
+            .is_some_match(as_output.datum.as_ref().map(|x| x.hash.as_ref()));
 
         MatchOutcome::fold_all_of([a, b, c, d].into_iter())
     }
@@ -499,7 +507,11 @@ fn iter_tx_assets(tx: &ParsedTx) -> impl Iterator<Item = &Multiasset> {
 }
 
 fn iter_tx_datums(tx: &ParsedTx) -> impl Iterator<Item = &[u8]> {
-    let a = tx.outputs.iter().map(|x| x.datum_hash.as_ref());
+    let a = tx
+        .outputs
+        .iter()
+        .flat_map(|x| &x.datum)
+        .map(|x| x.hash.as_ref());
 
     a
 }

src/filters/select/eval/testing.rs

@@ -1,4 +1,4 @@
-use utxorpc_spec::utxorpc::v1alpha::cardano::{metadatum, AuxData, Metadata, Metadatum, Tx};
+use pallas::interop::utxorpc::spec::cardano::{metadatum, AuxData, Datum, Metadata, Metadatum, Tx};
 
 use super::*;
 
@@ -17,6 +17,7 @@ pub fn multiasset_combo(policy_hex: &str, asset_prefix: &str) -> Multiasset {
                 mint_coin: 0,
             },
         ],
+        redeemer: None,
     }
 }
 
@@ -41,7 +42,10 @@ pub fn test_vectors() -> Vec<Tx> {
                 multiasset_combo("7eae28af2208be856f7a119668ae52a49b73725e326dc16579dcc373", "abc"),
                 multiasset_combo("1e349c9bdea19fd6c147626a5260bc44b71635f398b67c59881df209", "123")
             ],
-            datum_hash: hex::decode("923918e403bf43c34b4ef6b48eb2ee04babed17320d8d1b9ff9ad086e86f44ec").unwrap().into(),
+            datum: Datum{
+                hash:hex::decode("923918e403bf43c34b4ef6b48eb2ee04babed17320d8d1b9ff9ad086e86f44ec").unwrap().into(),
+                ..Default::default()
+            }.into(),
             ..Default::default()
         }],
         auxiliary: Some(AuxData {
@@ -64,10 +68,14 @@ pub fn test_vectors() -> Vec<Tx> {
                 "7eae28af2208be856f7a119668ae52a49b73725e326dc16579dcc373",
                 "abc",
             )],
-            datum_hash: hex::decode(
-                "923918e403bf43c34b4ef6b48eb2ee04babed17320d8d1b9ff9ad086e86f44ec",
-            )
-            .unwrap()
+            datum: Datum {
+                hash: hex::decode(
+                    "923918e403bf43c34b4ef6b48eb2ee04babed17320d8d1b9ff9ad086e86f44ec",
+                )
+                .unwrap()
+                .into(),
+                ..Default::default()
+            }
             .into(),
             ..Default::default()
         }],
@@ -89,7 +97,10 @@ pub fn test_vectors() -> Vec<Tx> {
             assets: vec![
                 multiasset_combo("1e349c9bdea19fd6c147626a5260bc44b71635f398b67c59881df209", "123")
             ],
-            datum_hash: hex::decode("923918e403bf43c34b4ef6b48eb2ee04babed17320d8d1b9ff9ad086e86f44ec").unwrap().into(),
+            datum: Datum{
+                hash:hex::decode("923918e403bf43c34b4ef6b48eb2ee04babed17320d8d1b9ff9ad086e86f44ec").unwrap().into(),
+                ..Default::default()
+            }.into(),
             ..Default::default()
         }],
         auxiliary: Some(AuxData {

src/filters/wasm_plugin.rs

@@ -43,7 +43,7 @@ impl Stage {
         let output = match output {
             serde_json::Value::Null => vec![],
             serde_json::Value::Array(x) => x.into_iter().map(Record::GenericJson).collect(),
-            x @ _ => vec![Record::GenericJson(x)],
+            x => vec![Record::GenericJson(x)],
         };
 
         Ok(output)

src/framework/mod.rs

@@ -10,8 +10,8 @@ use std::path::PathBuf;
 pub use crate::cursor::Config as CursorConfig;
 
 // we use UtxoRpc as our canonical representation of a parsed Tx
-pub use utxorpc_spec::utxorpc::v1alpha::cardano::Block as ParsedBlock;
-pub use utxorpc_spec::utxorpc::v1alpha::cardano::Tx as ParsedTx;
+pub use pallas::interop::utxorpc::spec::cardano::Block as ParsedBlock;
+pub use pallas::interop::utxorpc::spec::cardano::Tx as ParsedTx;
 
 // we use GenesisValues from Pallas as our ChainConfig
 pub use pallas::ledger::traverse::wellknown::GenesisValues;

src/sources/mithril.rs

@@ -221,7 +221,6 @@ fn read_blocks_with_config(
         .map_err(|_| WorkerError::Panic)
 }
 
-
 #[derive(Stage)]
 #[stage(name = "source", unit = "()", worker = "Worker")]
 pub struct Stage {

src/sources/u5c.rs

@@ -1,19 +1,17 @@
 use futures::StreamExt;
 use gasket::framework::*;
-
-use pallas::ledger::traverse::MultiEraBlock;
-use pallas::network::miniprotocols::Point;
 use serde::Deserialize;
 use tonic::transport::Channel;
 use tonic::Streaming;
 use tracing::{debug, error};
 
-use utxorpc_spec::utxorpc::v1alpha::sync::any_chain_block::Chain;
-use utxorpc_spec::utxorpc::v1alpha::sync::chain_sync_service_client::ChainSyncServiceClient;
-use utxorpc_spec::utxorpc::v1alpha::sync::follow_tip_response::Action;
-use utxorpc_spec::utxorpc::v1alpha::sync::{
+use pallas::interop::utxorpc::spec::sync::any_chain_block::Chain;
+use pallas::interop::utxorpc::spec::sync::follow_tip_response::Action;
+use pallas::interop::utxorpc::spec::sync::sync_service_client::SyncServiceClient;
+use pallas::interop::utxorpc::spec::sync::{
     BlockRef, DumpHistoryRequest, FollowTipRequest, FollowTipResponse,
 };
+use pallas::network::miniprotocols::Point;
 
 use crate::framework::*;
 
@@ -28,7 +26,7 @@ fn point_to_blockref(point: Point) -> Option<BlockRef> {
 }
 
 pub struct Worker {
-    client: ChainSyncServiceClient<Channel>,
+    client: SyncServiceClient<Channel>,
     stream: Option<Streaming<FollowTipResponse>>,
     intersect: Option<BlockRef>,
     max_items_per_page: u32,
@@ -57,17 +55,6 @@ impl Worker {
                                 }
                             }
                         }
-                        Chain::Raw(bytes) => {
-                            let block = MultiEraBlock::decode(bytes).or_panic()?;
-
-                            let evt = ChainEvent::Apply(
-                                Point::Specific(block.slot(), block.hash().to_vec()),
-                                Record::CborBlock(bytes.to_vec()),
-                            );
-
-                            stage.output.send(evt.into()).await.or_panic()?;
-                            stage.chain_tip.set(block.slot() as i64);
-                        }
                     }
                 }
             }
@@ -91,17 +78,6 @@ impl Worker {
                                 }
                             }
                         }
-                        Chain::Raw(bytes) => {
-                            let block = MultiEraBlock::decode(bytes).or_panic()?;
-
-                            let evt = ChainEvent::Undo(
-                                Point::Specific(block.slot(), block.hash().to_vec()),
-                                Record::CborBlock(bytes.to_vec()),
-                            );
-
-                            stage.output.send(evt.into()).await.or_panic()?;
-                            stage.chain_tip.set(block.slot() as i64);
-                        }
                     }
                 }
             }
@@ -183,7 +159,7 @@ impl gasket::framework::Worker<Stage> for Worker {
     async fn bootstrap(stage: &Stage) -> Result<Self, WorkerError> {
         debug!("connecting");
 
-        let client = ChainSyncServiceClient::connect(stage.config.url.clone())
+        let client = SyncServiceClient::connect(stage.config.url.clone())
             .await
             .or_panic()?;