Add upgrade test

.env

@@ -147,4 +147,6 @@ INTEGRATION_TEST_EXPECTED_BLOCK_HEIGHT=10
 INTEGRATION_TEST_HOST=localhost
 INTEGRATION_TEST_PROTO=http
 
-MARKETPLACE_SMOKE_TEST=false
+# Version of sequencer protocol we want to test. If this is set to
+# `03`, marketplace upgrade will be tested.
+INTEGRATION_TEST_SEQUENCER_VERSION=02

.github/workflows/test.yml

@@ -77,6 +77,7 @@ jobs:
       - name: Test Integration
         env:
           NEXTEST_PROFILE: integration
+          INTEGRATION_TEST_SEQUENCER_VERSION: 02
         run: cargo nextest run --locked --release --all-features --verbose --nocapture
         timeout-minutes: 40
 
@@ -89,7 +90,7 @@ jobs:
       - name: Test Marketplace Integration
         env:
           NEXTEST_PROFILE: integration
-          MARKETPLACE_SMOKE_TEST: true
+          INTEGRATION_TEST_SEQUENCER_VERSION: 03
         run: cargo nextest run --locked --release --all-features --verbose --nocapture
         timeout-minutes: 40
 

client/src/lib.rs

@@ -1,6 +1,6 @@
 use anyhow::Context;
 use async_std::task::sleep;
-use espresso_types::{FeeAccount, FeeAmount, FeeMerkleTree};
+use espresso_types::{FeeAccount, FeeAmount, FeeMerkleTree, Header};
 use ethers::types::Address;
 use jf_merkle_tree::{
     prelude::{MerkleProof, Sha3Node},
@@ -9,14 +9,12 @@ use jf_merkle_tree::{
 use std::time::Duration;
 use surf_disco::{
     error::ClientError,
-    http::convert::DeserializeOwned,
     socket::{Connection, Unsupported},
     Url,
 };
 use vbs::version::StaticVersion;
 
 pub type SequencerApiVersion = StaticVersion<0, 1>;
-// pub type EspressoClient = surf_disco::Client<ClientError, SequencerApiVersion>;
 
 #[derive(Clone, Debug)]
 pub struct SequencerClient(surf_disco::Client<ClientError, SequencerApiVersion>);
@@ -46,13 +44,13 @@ impl SequencerClient {
             .context("getting Espresso transaction count")
     }
 
-    /// Subscribe to a stream to Block Headers
-    pub async fn subscribe_headers<FromServer: DeserializeOwned>(
+    /// Subscribe to a stream of Block Headers
+    pub async fn subscribe_headers(
         &self,
-        connect: &str,
-    ) -> anyhow::Result<Connection<FromServer, Unsupported, ClientError, SequencerApiVersion>> {
+        height: u64,
+    ) -> anyhow::Result<Connection<Header, Unsupported, ClientError, SequencerApiVersion>> {
         self.0
-            .socket(connect)
+            .socket(&format!("availability/stream/headers/{height}"))
             .subscribe()
             .await
             .context("subscribing to Espresso headers")

justfile

@@ -9,7 +9,7 @@ demo *args:
 
 demo-native:
     cargo build --release
-    scripts/demo-native -D
+    scripts/demo-native
 
 demo-native-mp:
     cargo build --release

process-compose-mp.yml

@@ -73,8 +73,8 @@ processes:
     environment:
       - ESPRESSO_SUBMIT_TRANSACTIONS_PORT=$ESPRESSO_SUBMIT_TRANSACTIONS_PRIVATE_FALLBACK_PORT
       - ESPRESSO_SUBMIT_TRANSACTIONS_SUBMIT_URL=http://localhost:$ESPRESSO_FALLBACK_BUILDER_SERVER_PORT/txn_submit
-      - ESPRESSO_SUBMIT_TRANSACTIONS_MIN_NAMESPACE=1
-      - ESPRESSO_SUBMIT_TRANSACTIONS_MAX_NAMESPACE=3
+      - ESPRESSO_SUBMIT_TRANSACTIONS_MIN_NAMESPACE=4
+      - ESPRESSO_SUBMIT_TRANSACTIONS_MAX_NAMESPACE=6
     depends_on:
       sequencer0:
         condition: process_healthy
@@ -93,8 +93,8 @@ processes:
     environment:
       - ESPRESSO_SUBMIT_TRANSACTIONS_PORT=$ESPRESSO_SUBMIT_TRANSACTIONS_PRIVATE_RESERVE_PORT
       - ESPRESSO_SUBMIT_TRANSACTIONS_SUBMIT_URL=http://localhost:$ESPRESSO_RESERVE_BUILDER_SERVER_PORT/txn_submit
-      - ESPRESSO_SUBMIT_TRANSACTIONS_MIN_NAMESPACE=4
-      - ESPRESSO_SUBMIT_TRANSACTIONS_MAX_NAMESPACE=6
+      - ESPRESSO_SUBMIT_TRANSACTIONS_MIN_NAMESPACE=1
+      - ESPRESSO_SUBMIT_TRANSACTIONS_MAX_NAMESPACE=3
     depends_on:
       sequencer0:
         condition: process_healthy

sequencer/src/bin/espresso-bridge.rs

@@ -186,9 +186,7 @@ async fn deposit(opt: Deposit) -> anyhow::Result<()> {
 
     // Wait for Espresso to catch up to the L1.
     let espresso_height = espresso.get_height().await?;
-    let mut headers = espresso
-        .subscribe_headers(&format!("availability/stream/headers/{espresso_height}"))
-        .await?;
+    let mut headers = espresso.subscribe_headers(espresso_height).await?;
     let espresso_block = loop {
         let header: Header = match headers.next().await.context("header stream ended")? {
             Ok(header) => header,

tests/Cargo.toml

@@ -6,8 +6,8 @@ edition = { workspace = true }
 publish = false
 
 [[test]]
-name = "test_demo_native"
-path = "demo_native.rs"
+name = "integration"
+path = "main.rs"
 
 [dev-dependencies]
 anyhow = { workspace = true }
@@ -19,3 +19,4 @@ ethers = { workspace = true }
 futures = { workspace = true }
 reqwest = { workspace = true, features = ["json", "blocking"] }
 surf-disco = { workspace = true }
+vbs = { workspace = true }