merge base

.github/workflows/test-e2e.yml

@@ -11,7 +11,9 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 10
     steps:
-    - uses: actions/checkout@v4
+    - name: Checkout code
+      uses: actions/checkout@v4
+
     - uses: actions/setup-go@v5
       with:
         go-version: 1.22.x
@@ -41,24 +43,15 @@ jobs:
           - "fork12-validium"
           - "fork12-rollup"
     steps:
-    - uses: actions/checkout@v4
-
-    - name: Checkout kurtosis-cdk repository
+    - name: Checkout Code
       uses: actions/checkout@v4
-      with:
-        go-version: ${{ matrix.go-version }}
-      env:
-        GOARCH: ${{ matrix.goarch }}
-
-    - name: Build Docker
-      run: make build-docker
 
-    - name: Checkout kurtosis-cdk
+    - name: Checkout Kurtosis CDK
       uses: actions/checkout@v4
       with:
         repository: 0xPolygon/kurtosis-cdk
         path: kurtosis-cdk
-        ref: v0.2.24
+        ref: v0.2.25
 
     - name: Install Kurtosis CDK tools
       uses: ./kurtosis-cdk/.github/actions/setup-kurtosis-cdk

config/config_render.go

@@ -63,7 +63,7 @@ func (c *ConfigRender) Merge() (string, error) {
 		dataToml := c.convertVarsToStrings(data.Content)
 		err := k.Load(rawbytes.Provider([]byte(dataToml)), toml.Parser())
 		if err != nil {
-			log.Errorf("error loading file %s. Err:%v.FileData: %v", data.Name, err, dataToml)
+			log.Errorf("error loading file %s. Err:%v.", data.Name, err)
 			return "", fmt.Errorf("fail to load  converted template %s to toml. Err: %w", data.Name, err)
 		}
 	}

crates/cdk/build.rs

@@ -2,10 +2,11 @@ use regex::Regex;
 use reqwest::blocking::get;
 use std::env;
 use std::fs::File;
-use std::io::Write;
+use std::io::{self, Write};
 use std::path::Path;
 use std::path::PathBuf;
 use std::process::Command;
+use serde_json::Value;
 
 fn main() {
     let _ = build_versions();
@@ -55,34 +56,28 @@ fn main() {
 }
 
 // build_versions retrieves the versions from the Starlark file and embeds them in the binary.
-fn build_versions() -> std::io::Result<()> {
-    // Retrieve the contents of the file from the URL
+fn build_versions() -> io::Result<()> {
+    // URL of the Starlark file
     let url = "https://raw.githubusercontent.com/0xPolygon/kurtosis-cdk/refs/heads/main/input_parser.star";
+
+    // Download the file content
     let response = get(url).expect("Failed to send request");
     let content = response.text().expect("Failed to read response text");
 
-    // Write the contents to a file
-    let out_dir = std::env::var("OUT_DIR").unwrap();
-    let dest_path = Path::new(&out_dir).join("input_parser.star");
-    let mut file = File::create(&dest_path)?;
-    file.write_all(content.as_bytes())?;
-
-    // Get the corresponding lines from the contents of the starlark file
-    let versions = content
+    // Extract the relevant lines (skip the first 30 lines, take the next 15)
+    let raw_versions = content
         .lines()
         .skip(30)
         .take(15)
         .collect::<Vec<&str>>()
         .join("\n");
 
-    // Replace the string DEFAULT_IMAGES = from the versions string
-    let versions = versions.replace("DEFAULT_IMAGES = ", "");
+    // Remove the declaration `DEFAULT_IMAGES = `
+    let raw_versions = raw_versions.replace("DEFAULT_IMAGES = ", "");
 
-    // Remove all comments to the end of the line using a regexp
-    let re = Regex::new(r"\s#\s.*\n").unwrap();
-    let versions = re.replace_all(&versions, "");
-    // Replace the trailing comma on the last line
-    let versions = versions.replace(", }", " }");
+    // Clean up the content by removing comments and unnecessary spaces
+    let re_comments = Regex::new(r"#.*$").unwrap(); // Regex to remove comments
+    let re_trailing_commas = Regex::new(r",(\s*})").unwrap(); // Regex to fix trailing commas
 
     // The versions string is a JSON object we can parse
     let versions_json: serde_json::Value = match serde_json::from_str(&versions) {
@@ -93,13 +88,26 @@ fn build_versions() -> std::io::Result<()> {
         }
     };
 
-    // Write the versions to a file
+    // Fix improperly placed trailing commas
+    let cleaned_versions = re_trailing_commas.replace_all(&cleaned_versions, "$1");
+
+    // Attempt to parse the cleaned content as JSON
+    let versions_json: Value = match serde_json::from_str(&cleaned_versions) {
+        Ok(json) => json,
+        Err(e) => {
+            eprintln!("Failed to parse JSON: {}", e); // Print the error
+            eprintln!("Input string was: {}", cleaned_versions); // Print the input causing the error
+            return Err(io::Error::new(io::ErrorKind::InvalidData, "JSON parsing failed"));
+        }
+    };
+
+    // Define the output file path for the JSON
     let dest_path = Path::new(".").join("versions.json");
     let mut file = File::create(&dest_path)?;
     file.write_all(
         format!(
             "{}\n",
-            serde_json::to_string_pretty(&versions_json).unwrap()
+            serde_json::to_string_pretty(&versions_json).unwrap() // Pretty-print JSON to the file
         )
         .as_bytes(),
     )?;

reorgdetector/reorgdetector.go

@@ -159,10 +159,11 @@ func (rd *ReorgDetector) detectReorgInTrackedList(ctx context.Context) error {
 					// and hashes matches. If higher than finalized block, we assume a reorg still might happen.
 					if hdr.Num <= lastFinalisedBlock.Number.Uint64() {
 						hdrs.removeRange(hdr.Num, hdr.Num)
-					}
-					if err := rd.removeTrackedBlockRange(id, hdr.Num, hdr.Num); err != nil {
-						return fmt.Errorf("error removing blocks from DB for subscriber %s between blocks %d and %d: %w",
-							id, hdr.Num, hdr.Num, err)
+
+						if err := rd.removeTrackedBlockRange(id, hdr.Num, hdr.Num); err != nil {
+							return fmt.Errorf("error removing blocks from DB for subscriber %s between blocks %d and %d: %w",
+								id, hdr.Num, hdr.Num, err)
+						}
 					}
 
 					continue

reorgdetector/reorgdetector_db.go

@@ -70,7 +70,7 @@ func (rd *ReorgDetector) saveTrackedBlock(id string, b header) error {
 // updateTrackedBlocksDB updates the tracked blocks for a subscriber in db
 func (rd *ReorgDetector) removeTrackedBlockRange(id string, fromBlock, toBlock uint64) error {
 	_, err := rd.db.Exec(
-		"DELETE FROM tracked_block WHERE num >= $1 AND NUM <= 2 AND subscriber_id = $3;",
+		"DELETE FROM tracked_block WHERE num >= $1 AND num <= $2 AND subscriber_id = $3;",
 		fromBlock, toBlock, id,
 	)
 	return err

reorgdetector/reorgdetector_test.go

@@ -2,14 +2,18 @@ package reorgdetector
 
 import (
 	"context"
+	big "math/big"
 	"path"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	cdktypes "github.com/0xPolygon/cdk/config/types"
 	common "github.com/ethereum/go-ethereum/common"
+	types "github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethclient/simulated"
+	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/stretchr/testify/require"
 )
 
@@ -158,3 +162,98 @@ func TestNotSubscribed(t *testing.T) {
 	err = reorgDetector.AddBlockToTrack(context.Background(), "foo", 1, common.Hash{})
 	require.True(t, strings.Contains(err.Error(), "is not subscribed"))
 }
+
+func TestDetectReorgs(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	syncerID := "test-syncer"
+	trackedBlock := &types.Header{Number: big.NewInt(9)}
+
+	t.Run("Block not finalized", func(t *testing.T) {
+		t.Parallel()
+
+		lastFinalizedBlock := &types.Header{Number: big.NewInt(8)}
+		client := NewEthClientMock(t)
+		client.On("HeaderByNumber", ctx, big.NewInt(int64(rpc.FinalizedBlockNumber))).Return(lastFinalizedBlock, nil)
+		client.On("HeaderByNumber", ctx, trackedBlock.Number).Return(trackedBlock, nil)
+
+		testDir := path.Join(t.TempDir(), "reorgdetectorTestDetectReorgs.sqlite")
+		reorgDetector, err := New(client, Config{DBPath: testDir, CheckReorgsInterval: cdktypes.NewDuration(time.Millisecond * 100)})
+		require.NoError(t, err)
+
+		_, err = reorgDetector.Subscribe(syncerID)
+		require.NoError(t, err)
+		require.NoError(t, reorgDetector.AddBlockToTrack(ctx, syncerID, trackedBlock.Number.Uint64(), trackedBlock.Hash()))
+
+		require.NoError(t, reorgDetector.detectReorgInTrackedList(ctx))
+
+		trackedBlocks, err := reorgDetector.getTrackedBlocks()
+		require.NoError(t, err)
+		require.Equal(t, 1, len(trackedBlocks))
+
+		syncerTrackedBlocks, ok := trackedBlocks[syncerID]
+		require.True(t, ok)
+		require.Equal(t, 1, syncerTrackedBlocks.len())
+	})
+
+	t.Run("Block finalized", func(t *testing.T) {
+		t.Parallel()
+
+		lastFinalizedBlock := trackedBlock
+		client := NewEthClientMock(t)
+		client.On("HeaderByNumber", ctx, big.NewInt(int64(rpc.FinalizedBlockNumber))).Return(lastFinalizedBlock, nil)
+
+		testDir := path.Join(t.TempDir(), "reorgdetectorTestDetectReorgs.sqlite")
+		reorgDetector, err := New(client, Config{DBPath: testDir, CheckReorgsInterval: cdktypes.NewDuration(time.Millisecond * 100)})
+		require.NoError(t, err)
+
+		_, err = reorgDetector.Subscribe(syncerID)
+		require.NoError(t, err)
+		require.NoError(t, reorgDetector.AddBlockToTrack(ctx, syncerID, trackedBlock.Number.Uint64(), trackedBlock.Hash()))
+
+		require.NoError(t, reorgDetector.detectReorgInTrackedList(ctx))
+
+		trackedBlocks, err := reorgDetector.getTrackedBlocks()
+		require.NoError(t, err)
+		require.Equal(t, 0, len(trackedBlocks))
+	})
+
+	t.Run("Reorg happened", func(t *testing.T) {
+		t.Parallel()
+
+		lastFinalizedBlock := &types.Header{Number: big.NewInt(5)}
+		reorgedTrackedBlock := &types.Header{Number: trackedBlock.Number, Extra: []byte("reorged")} // Different hash
+
+		client := NewEthClientMock(t)
+		client.On("HeaderByNumber", ctx, big.NewInt(int64(rpc.FinalizedBlockNumber))).Return(lastFinalizedBlock, nil)
+		client.On("HeaderByNumber", ctx, trackedBlock.Number).Return(reorgedTrackedBlock, nil)
+
+		testDir := path.Join(t.TempDir(), "reorgdetectorTestDetectReorgs.sqlite")
+		reorgDetector, err := New(client, Config{DBPath: testDir, CheckReorgsInterval: cdktypes.NewDuration(time.Millisecond * 100)})
+		require.NoError(t, err)
+
+		subscription, err := reorgDetector.Subscribe(syncerID)
+		require.NoError(t, err)
+
+		var wg sync.WaitGroup
+
+		wg.Add(1)
+		go func() {
+			<-subscription.ReorgedBlock
+			subscription.ReorgProcessed <- true
+
+			wg.Done()
+		}()
+
+		require.NoError(t, reorgDetector.AddBlockToTrack(ctx, syncerID, trackedBlock.Number.Uint64(), trackedBlock.Hash()))
+
+		require.NoError(t, reorgDetector.detectReorgInTrackedList(ctx))
+
+		wg.Wait() // we wait here to make sure the reorg is processed
+
+		trackedBlocks, err := reorgDetector.getTrackedBlocks()
+		require.NoError(t, err)
+		require.Equal(t, 0, len(trackedBlocks)) // shouldn't be any since a reorg happened on that block
+	})
+}