feat(phase-1 validations): add support for ShelleyMA

Cargo.toml

@@ -12,7 +12,7 @@ authors = ["Santiago Carmuega <santiago@carmuega.me>"]
 
 
 [dependencies]
-pallas = { git = "https://github.com/txpipe/pallas.git", features = ["unstable"] }
+pallas = { git = "https://github.com/txpipe/pallas.git", branch = "dev/shelleyMA-phase-1-validations", features = ["unstable"] }
 
 gasket = { version = "^0.5", features = ["derive"] }
 # gasket = { path = "../../construkts/gasket-rs/gasket", features = ["derive"] }

examples/sync-mainnet/dolos.toml

@@ -1,6 +1,7 @@
 [upstream]
 peer_address = "relays-new.cardano-mainnet.iohk.io:3001"
 network_magic = 764824073
+network_id = 1
 
 [rolldb]
 path = "./tmp/rolldb"

examples/sync-preprod/dolos.toml

@@ -1,6 +1,7 @@
 [upstream]
 peer_address = "preprod-node.world.dev.cardano.org:30000"
 network_magic = 1
+network_id = 0
 
 [rolldb]
 path = "./tmp/rolldb"

examples/sync-preview/dolos.toml

@@ -1,6 +1,7 @@
 [upstream]
 peer_address = "preview-node.world.dev.cardano.org:30002"
 network_magic = 2
+network_id = 0
 
 [rolldb]
 path = "./tmp/rolldb"

src/storage/applydb/mod.rs

@@ -2,7 +2,7 @@ pub mod genesis;
 
 use pallas::{
     applying::{
-        types::{ByronProtParams, Environment, MultiEraProtParams, UTxOs},
+        types::{ByronProtParams, Environment, FeePolicy, MultiEraProtParams, UTxOs},
         validate,
     },
     codec::utils::CborWrap,
@@ -299,7 +299,12 @@ impl ApplyDB {
         Ok(dbval.map(|x| x.0))
     }
 
-    pub fn apply_block(&mut self, cbor: &[u8], prot_magic: &u32) -> Result<(), Error> {
+    pub fn apply_block(
+        &mut self,
+        cbor: &[u8],
+        prot_magic: &u32,
+        network_id: &u8,
+    ) -> Result<(), Error> {
         let block = MultiEraBlock::decode(cbor).map_err(|_| Error::Cbor)?;
         let slot = block.slot();
         let hash = block.hash();
@@ -333,18 +338,22 @@ impl ApplyDB {
         }
 
         for tx in txs.iter() {
-            if tx.era() == Era::Byron {
-                match self.get_inputs(tx, &batch) {
+            match tx.era() {
+                Era::Byron | Era::Shelley | Era::Allegra | Era::Mary => match self
+                    .get_inputs(tx, &batch)
+                {
                     Ok(inputs) => {
                         let utxos: UTxOs = Self::mk_utxo(&inputs);
-                        let env: Environment = Self::mk_environment(&block, prot_magic)?;
+                        let env: Environment =
+                            Self::mk_environment(&block, prot_magic, network_id)?;
                         match validate(tx, &utxos, &env) {
                             Ok(()) => (),
                             Err(err) => warn!("Transaction validation failed ({:?})", err),
                         }
                     }
                     Err(err) => warn!("Skipping validation ({:?})", err),
-                }
+                },
+                _ => (),
             }
         }
 
@@ -400,40 +409,56 @@ impl ApplyDB {
         utxos
     }
 
-    fn mk_environment(block: &MultiEraBlock, prot_magic: &u32) -> Result<Environment, Error> {
+    fn mk_environment(
+        block: &MultiEraBlock,
+        prot_magic: &u32,
+        network_id: &u8,
+    ) -> Result<Environment, Error> {
         if block.era() == Era::Byron {
             let slot: u64 = block.header().slot();
             if slot <= 322876 {
                 // These are the genesis values.
                 Ok(Environment {
                     prot_params: MultiEraProtParams::Byron(ByronProtParams {
-                        min_fees_const: 155381,
-                        min_fees_factor: 44,
+                        fee_policy: FeePolicy {
+                            summand: 155381,
+                            multiplier: 44,
+                        },
                         max_tx_size: 4096,
                     }),
                     prot_magic: *prot_magic,
+                    block_slot: slot,
+                    network_id: *network_id,
                 })
             } else if slot > 322876 && slot <= 1784895 {
                 // Block hash were the update proposal was submitted:
                 // 850805044e0df6c13ced2190db7b11489672b0225d478a35a6db71fbfb33afc0
                 Ok(Environment {
                     prot_params: MultiEraProtParams::Byron(ByronProtParams {
-                        min_fees_const: 155381,
-                        min_fees_factor: 44,
+                        fee_policy: FeePolicy {
+                            summand: 155381,
+                            multiplier: 44,
+                        },
                         max_tx_size: 65536,
                     }),
                     prot_magic: *prot_magic,
+                    block_slot: slot,
+                    network_id: *network_id,
                 })
             } else {
                 // Block hash were the update proposal was submitted:
                 // d798a8d617b25fc6456ffe2d90895a2c15a7271b671dab2d18d46f3d0e4ef495
                 Ok(Environment {
                     prot_params: MultiEraProtParams::Byron(ByronProtParams {
-                        min_fees_const: 155381,
-                        min_fees_factor: 44,
+                        fee_policy: FeePolicy {
+                            summand: 155381,
+                            multiplier: 44,
+                        },
                         max_tx_size: 8192,
                     }),
                     prot_magic: *prot_magic,
+                    block_slot: slot,
+                    network_id: *network_id,
                 })
             }
         } else {
@@ -564,7 +589,9 @@ mod tests {
                 }
             }
 
-            db.apply_block(&cbor, &764824073).unwrap(); // This is mainnet's protocol magic number.
+            db.apply_block(&cbor, &764824073, &1).unwrap();
+            //                                 ^ mainnet network ID
+            //                     ^ mainnet protocol magic number
 
             for tx in block.txs() {
                 for input in tx.consumes() {

src/sync/ledger.rs

@@ -13,6 +13,7 @@ pub struct Stage {
     ledger: ApplyDB,
     genesis: GenesisFile,
     prot_magic: u32,
+    network_id: u8,
 
     pub upstream: UpstreamPort,
 
@@ -24,13 +25,13 @@ pub struct Stage {
 }
 
 impl Stage {
-    pub fn new(ledger: ApplyDB, genesis: GenesisFile, prot_magic: u64) -> Self {
+    pub fn new(ledger: ApplyDB, genesis: GenesisFile, prot_magic: u64, network_id: u8) -> Self {
         Self {
             ledger,
             genesis,
             prot_magic: prot_magic as u32,
+            network_id,
             upstream: Default::default(),
-            // downstream: Default::default(),
             block_count: Default::default(),
             wal_count: Default::default(),
         }
@@ -60,7 +61,7 @@ impl gasket::framework::Worker<Stage> for Worker {
                 info!(slot, "applying block");
                 stage
                     .ledger
-                    .apply_block(cbor, &stage.prot_magic)
+                    .apply_block(cbor, &stage.prot_magic, &stage.network_id)
                     .or_panic()?;
             }
             RollEvent::Undo(slot, _, cbor) => {

src/sync/mod.rs

@@ -19,6 +19,7 @@ pub mod roll;
 pub struct Config {
     peer_address: String,
     network_magic: u64,
+    network_id: u8,
 }
 
 fn define_gasket_policy(config: &Option<gasket::retries::Policy>) -> gasket::runtime::Policy {
@@ -70,7 +71,7 @@ pub fn pipeline(
     let mut roll = roll::Stage::new(wal, cursor_chain, cursor_ledger);
 
     let mut chain = chain::Stage::new(chain);
-    let mut ledger = ledger::Stage::new(ledger, genesis, config.network_magic);
+    let mut ledger = ledger::Stage::new(ledger, genesis, config.network_magic, config.network_id);
 
     let (to_roll, from_pull) = gasket::messaging::tokio::mpsc_channel(50);
     pull.downstream.connect(to_roll);