chore: update Pallas to v0.27 (#791)

Cargo.toml

@@ -22,16 +22,16 @@ u5c = ["tonic"]
 # kafka = auto feature flag
 
 [dependencies]
-# pallas = "0.21.0"
+pallas = "0.27.0"
 # pallas = { path = "../pallas/pallas" }
-pallas = { git = "https://github.com/txpipe/pallas" }
+# pallas = { git = "https://github.com/txpipe/pallas" }
 
 gasket = { version = "^0.7", features = ["derive"] }
 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 = { version = "^0.1" }
+utxorpc-spec = { version = "^0.5" }
 
 hex = "0.4.3"
 net2 = "0.2.37"
@@ -68,7 +68,7 @@ google-cloud-googleapis = { version = "0.10.0", optional = true }
 google-cloud-default = { version = "0.4.0", optional = true, features = ["pubsub"] }
 r2d2_redis = { version = "0.14.0", optional = true }
 jsonwebtoken = { version = "8.3.0", optional = true }
-tonic = { version = "0.9.2", features = ["tls", "tls-roots"], optional = true }
+tonic = { version = "0.11", features = ["tls", "tls-roots"], optional = true }
 futures = { version = "0.3.28", optional = true }
 sqlx = { version = "0.7", features = ["runtime-tokio", "tls-native-tls", "any", "sqlite", "postgres"], optional = true }
 aws-config = { version = "^1.1", optional = true }

src/filters/legacy_v1/crawl.rs

@@ -86,7 +86,7 @@ impl EventWriter<'_> {
         }
 
         for redeemer in tx.redeemers() {
-            self.append_from(self.to_plutus_redeemer_record(redeemer))?;
+            self.append_from(self.to_plutus_redeemer_record(&redeemer))?;
         }
 
         for datum in tx.plutus_data() {

src/filters/legacy_v1/map.rs

@@ -1,4 +1,6 @@
 use gasket::framework::AsWorkError;
+use pallas::codec::utils::Nullable;
+use pallas::ledger::primitives::conway;
 use serde_json::{json, Value as JsonValue};
 use std::collections::HashMap;
 use std::ops::Deref as _;
@@ -14,7 +16,7 @@ use pallas::ledger::primitives::{
 };
 use pallas::ledger::traverse::{
     ComputeHash, MultiEraAsset, MultiEraBlock, MultiEraCert, MultiEraInput, MultiEraOutput,
-    MultiEraTx, OriginalHash,
+    MultiEraRedeemer, MultiEraTx, OriginalHash,
 };
 use pallas::network::miniprotocols::Point;
 use pallas::{codec::utils::KeepRaw, crypto::hash::Hash};
@@ -106,19 +108,19 @@ fn relay_to_string(relay: &Relay) -> String {
     match relay {
         Relay::SingleHostAddr(port, ipv4, ipv6) => {
             let ip = match (ipv6, ipv4) {
-                (None, None) => "".to_string(),
-                (_, Some(x)) => ip_string_from_bytes(x.as_ref()),
-                (Some(x), _) => ip_string_from_bytes(x.as_ref()),
+                (_, Nullable::Some(x)) => ip_string_from_bytes(x.as_ref()),
+                (Nullable::Some(x), _) => ip_string_from_bytes(x.as_ref()),
+                _ => "".to_string(),
             };
 
             match port {
-                Some(port) => format!("{ip}:{port}"),
-                None => ip,
+                Nullable::Some(port) => format!("{ip}:{port}"),
+                _ => ip,
             }
         }
         Relay::SingleHostName(port, host) => match port {
-            Some(port) => format!("{host}:{port}"),
-            None => host.clone(),
+            Nullable::Some(port) => format!("{host}:{port}"),
+            _ => host.clone(),
         },
         Relay::MultiHostName(host) => host.clone(),
     }
@@ -396,18 +398,20 @@ impl EventWriter<'_> {
         }
     }
 
-    pub fn to_plutus_redeemer_record(&self, redeemer: &alonzo::Redeemer) -> PlutusRedeemerRecord {
+    pub fn to_plutus_redeemer_record(&self, redeemer: &MultiEraRedeemer) -> PlutusRedeemerRecord {
         PlutusRedeemerRecord {
-            purpose: match redeemer.tag {
-                alonzo::RedeemerTag::Spend => "spend".to_string(),
-                alonzo::RedeemerTag::Mint => "mint".to_string(),
-                alonzo::RedeemerTag::Cert => "cert".to_string(),
-                alonzo::RedeemerTag::Reward => "reward".to_string(),
+            purpose: match redeemer.tag() {
+                conway::RedeemerTag::Spend => "spend".to_string(),
+                conway::RedeemerTag::Mint => "mint".to_string(),
+                conway::RedeemerTag::Cert => "cert".to_string(),
+                conway::RedeemerTag::Reward => "reward".to_string(),
+                conway::RedeemerTag::Vote => "vote".to_string(),
+                conway::RedeemerTag::Propose => "propose".to_string(),
             },
-            ex_units_mem: redeemer.ex_units.mem,
-            ex_units_steps: redeemer.ex_units.steps,
-            input_idx: redeemer.index,
-            plutus_data: redeemer.data.to_json(),
+            ex_units_mem: redeemer.ex_units().mem,
+            ex_units_steps: redeemer.ex_units().steps,
+            input_idx: redeemer.index(),
+            plutus_data: redeemer.data().to_json(),
         }
     }
 
@@ -476,8 +480,14 @@ impl EventWriter<'_> {
                 reward_account: reward_account.to_hex(),
                 pool_owners: pool_owners.iter().map(|p| p.to_hex()).collect(),
                 relays: relays.iter().map(relay_to_string).collect(),
-                pool_metadata: pool_metadata.as_ref().map(|m| m.url.clone()),
-                pool_metadata_hash: pool_metadata.as_ref().map(|m| m.hash.clone().to_hex()),
+                pool_metadata: match pool_metadata {
+                    Nullable::Some(x) => Some(x.url.clone()),
+                    _ => None,
+                },
+                pool_metadata_hash: match pool_metadata {
+                    Nullable::Some(x) => Some(x.hash.clone().to_hex()),
+                    _ => None,
+                },
             },
             Certificate::PoolRetirement(pool, epoch) => EventData::PoolRetirement {
                 pool: pool.to_hex(),

src/filters/parse_cbor.rs

@@ -3,17 +3,32 @@
 use gasket::framework::*;
 use serde::Deserialize;
 
-use pallas::interop::utxorpc as interop;
+use pallas::interop::utxorpc::{self as interop};
 use pallas::ledger::traverse as trv;
 
 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>> {
+        None
+    }
+}
+
 #[derive(Default, Stage)]
 #[stage(name = "filter-parse-cbor", unit = "ChainEvent", worker = "Worker")]
 pub struct Stage {
     pub input: FilterInputPort,
     pub output: FilterOutputPort,
 
+    mapper: interop::Mapper<NoOpContext>,
+
     #[metric]
     ops_count: gasket::metrics::Counter,
 }
@@ -31,7 +46,7 @@ gasket::impl_mapper!(|_worker: Worker, stage: Stage, unit: ChainEvent| => {
     let output = unit.clone().try_map_record(|r| match r {
         Record::CborTx(cbor) => {
             let tx = trv::MultiEraTx::decode(&cbor).or_panic()?;
-            let tx = interop::map_tx(&tx);
+            let tx = stage.mapper.map_tx(&tx);
             Ok(Record::ParsedTx(tx))
         }
         x => Ok(x),

src/filters/select/eval/mod.rs

@@ -2,7 +2,9 @@ use std::{ops::Deref, str::FromStr};
 
 use serde::{Deserialize, Serialize};
 use tracing::warn;
-use utxorpc::spec::cardano::{Asset, AuxData, Metadata, Metadatum, Multiasset, TxInput, TxOutput};
+use utxorpc_spec::utxorpc::v1alpha::cardano::{
+    Asset, AuxData, Metadata, Metadatum, Multiasset, TxInput, TxOutput,
+};
 
 use crate::framework::*;
 
@@ -216,7 +218,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::cardano::metadatum::Metadatum::Text(subject)) => {
+            Some(utxorpc_spec::utxorpc::v1alpha::cardano::metadatum::Metadatum::Text(subject)) => {
                 self.is_match(subject.as_str())
             }
             _ => MatchOutcome::Negative,

src/filters/select/eval/testing.rs

@@ -1,4 +1,4 @@
-use utxorpc::spec::cardano::{metadatum, AuxData, Metadata, Metadatum, Tx};
+use utxorpc_spec::utxorpc::v1alpha::cardano::{metadatum, AuxData, Metadata, Metadatum, Tx};
 
 use super::*;
 

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::cardano::Block as ParsedBlock;
-pub use utxorpc::spec::cardano::Tx as ParsedTx;
+pub use utxorpc_spec::utxorpc::v1alpha::cardano::Block as ParsedBlock;
+pub use utxorpc_spec::utxorpc::v1alpha::cardano::Tx as ParsedTx;
 
 // we use GenesisValues from Pallas as our ChainConfig
 pub use pallas::ledger::traverse::wellknown::GenesisValues;
@@ -174,7 +174,10 @@ impl ChainEvent {
         }
     }
 
-    pub fn try_map_record<E>(self, f: fn(Record) -> Result<Record, E>) -> Result<Self, E> {
+    pub fn try_map_record<E, F>(self, f: F) -> Result<Self, E>
+    where
+        F: FnOnce(Record) -> Result<Record, E>,
+    {
         let out = match self {
             Self::Apply(p, x) => Self::Apply(p, f(x)?),
             Self::Undo(p, x) => Self::Undo(p, f(x)?),

src/sources/u5c.rs

@@ -8,10 +8,12 @@ use tonic::transport::Channel;
 use tonic::Streaming;
 use tracing::{debug, error};
 
-use utxorpc::spec::sync::any_chain_block::Chain;
-use utxorpc::spec::sync::chain_sync_service_client::ChainSyncServiceClient;
-use utxorpc::spec::sync::follow_tip_response::Action;
-use utxorpc::spec::sync::{BlockRef, DumpHistoryRequest, FollowTipRequest, FollowTipResponse};
+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::{
+    BlockRef, DumpHistoryRequest, FollowTipRequest, FollowTipResponse,
+};
 
 use crate::framework::*;